diff --git a/src/__tests__/dashboard-summary-read-plan-guard.test.ts b/src/__tests__/dashboard-summary-read-plan-guard.test.ts index b1b9a14e1..ce7035eb5 100644 --- a/src/__tests__/dashboard-summary-read-plan-guard.test.ts +++ b/src/__tests__/dashboard-summary-read-plan-guard.test.ts @@ -85,7 +85,7 @@ describe("dashboard summary reads — sparkline rollup buckets", () => { // defeat statement reuse. const sql = code(READS); const casts = sql.match(/\}::"measurement_type"\[\]/g) ?? []; - expect(casts.length).toBe(2); + expect(casts.length).toBe(3); expect(sql).not.toMatch(/\$queryRawUnsafe/); }); }); diff --git a/src/__tests__/dense-measurement-read-guard.test.ts b/src/__tests__/dense-measurement-read-guard.test.ts index 75c083e73..07c9d441b 100644 --- a/src/__tests__/dense-measurement-read-guard.test.ts +++ b/src/__tests__/dense-measurement-read-guard.test.ts @@ -134,6 +134,8 @@ const ALLOWED_SQL: Record = { "lib/rollups/measurement-coverage.ts::1": "Selects from `unnest` of the caller's type list; `measurements` is read inside `EXISTS`, one row per type at most.", "lib/dashboard/summary-reads.ts::1": + "Selects from `unnest` of the type list; `measurements` is read only inside COUNT scalar subqueries, before the fold window for a type the DAY rollups cover.", + "lib/dashboard/summary-reads.ts::2": "Selects from `generate_series` of the streak window's local days; `measurements` is read inside `EXISTS`, one row per day at most.", }; diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 4deacda2d..da045a6ed 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -176,7 +176,9 @@ describe("measurement write tail", () => { const missing = handRolled.filter((file) => { const src = readFileSync(join(SRC, file), "utf8"); return !( - /recomputeBucketsForMeasurement|afterMeasurementMutation/.test(src) && + /recomputeBucketsFor(Measurement|Keys)|afterMeasurementMutation/.test( + src, + ) && /invalidateStatusInsightsForTypes|afterMeasurementMutation/.test(src) ); }); diff --git a/src/__tests__/no-server-actions-guard.test.ts b/src/__tests__/no-server-actions-guard.test.ts new file mode 100644 index 000000000..8dd2b7700 --- /dev/null +++ b/src/__tests__/no-server-actions-guard.test.ts @@ -0,0 +1,37 @@ +/** + * HealthLog defines no Server Actions. + * + * The proxy refuses every request carrying `Next-Action` on the strength of + * that (`src/proxy.ts`): the only callers that send one are scanners. A module + * that declares `"use server"` would make the refusal break a real feature, so + * adding one has to come with revisiting it. This guard is the reminder. + */ +import { readdirSync, readFileSync, statSync } from "node:fs"; +import { join, relative } from "node:path"; +import { describe, expect, it } from "vitest"; + +const SRC = join(__dirname, ".."); + +function sourceFiles(dir: string): string[] { + const out: string[] = []; + for (const name of readdirSync(dir)) { + const path = join(dir, name); + if (name === "generated" || name === "__tests__") continue; + if (statSync(path).isDirectory()) out.push(...sourceFiles(path)); + else if (/\.(ts|tsx)$/.test(name)) out.push(path); + } + return out; +} + +describe("no Server Actions", () => { + it('has no module or function that declares "use server"', () => { + const files = sourceFiles(SRC); + expect(files.length).toBeGreaterThan(100); + const declaring = files + .filter((file) => + /^\s*["']use server["'];?\s*$/m.test(readFileSync(file, "utf8")), + ) + .map((file) => relative(SRC, file)); + expect(declaring).toEqual([]); + }); +}); diff --git a/src/__tests__/proxy-server-action.test.ts b/src/__tests__/proxy-server-action.test.ts new file mode 100644 index 000000000..a41e8a129 --- /dev/null +++ b/src/__tests__/proxy-server-action.test.ts @@ -0,0 +1,77 @@ +/** + * A Server Action call is refused at the proxy. + * + * HealthLog defines no Server Actions, so the only requests that carry + * `Next-Action` are scanners probing for them. Next used to answer each with + * an unstructured "Failed to find Server Action" at error level; the proxy now + * refuses it with a 404 and one structured `warn` line, and a request without + * the header is untouched. + */ +import { NextRequest } from "next/server"; +import { + afterEach, + beforeEach, + describe, + expect, + it, + vi, + type MockInstance, +} from "vitest"; + +vi.mock("@/lib/process-type", () => ({ shouldRunWeb: () => true })); + +import { proxy } from "../proxy"; + +let warn: MockInstance<(...args: unknown[]) => void>; + +beforeEach(() => { + warn = vi.spyOn(console, "warn").mockImplementation(() => {}); +}); + +afterEach(() => { + warn.mockRestore(); +}); + +function probe(path: string, headers: Record = {}) { + return new NextRequest(`https://health.example.test${path}`, { + method: "POST", + headers: { "content-type": "text/plain", ...headers }, + body: "[]", + }); +} + +describe("proxy on a Server Action call", () => { + it("refuses it with 404 and one warn line", async () => { + const response = proxy( + probe("/", { "next-action": "7f3a9c", "user-agent": "scanner" }), + ); + + expect(response.status).toBe(404); + expect(await response.json()).toEqual({ + data: null, + error: "Not found", + meta: { errorCode: "server_action.unsupported" }, + }); + expect(response.headers.get("X-Content-Type-Options")).toBe("nosniff"); + expect(warn).toHaveBeenCalledTimes(1); + expect(JSON.parse(String(warn.mock.calls[0]![0]))).toMatchObject({ + level: "warn", + action: { name: "proxy.server_action.refused" }, + http: { method: "POST", path: "/", status: 404, user_agent: "scanner" }, + }); + }); + + it("never logs a secret path segment the probe aimed at", () => { + proxy(probe("/api/withings/webhook/s3cr3t-value", { "next-action": "x" })); + + const line = String(warn.mock.calls[0]![0]); + expect(line).not.toContain("s3cr3t-value"); + }); + + it("leaves a request without the header alone", () => { + const response = proxy(probe("/api/health")); + + expect(response.status).not.toBe(404); + expect(warn).not.toHaveBeenCalled(); + }); +}); diff --git a/src/app/api/admin/central-codex/device-poll/route.ts b/src/app/api/admin/central-codex/device-poll/route.ts index b5e77028f..86aab4eed 100644 --- a/src/app/api/admin/central-codex/device-poll/route.ts +++ b/src/app/api/admin/central-codex/device-poll/route.ts @@ -8,6 +8,7 @@ import { prisma } from "@/lib/db"; import { auditLog } from "@/lib/auth/audit"; import { getClientIp, apiSuccess, apiError } from "@/lib/api-response"; import { annotate } from "@/lib/logging/context"; +import { forgetProviderHealth } from "@/lib/ai/provider-health-ledger"; /** * Operator-shared central Codex — device-code poll (admin only). @@ -91,6 +92,9 @@ export const POST = apiHandler(async (request: NextRequest) => { }); cookieStore.delete("central_codex_device"); + // A new central sign-in: every account's ledger rows describe the refused + // one, and the central connection is one credential shared by all of them. + await forgetProviderHealth({ providerType: "admin-codex" }); await auditLog("admin.central-codex.connected", { userId: user.id, diff --git a/src/app/api/auth/codex/device-poll/route.ts b/src/app/api/auth/codex/device-poll/route.ts index 1853de300..47f9987b0 100644 --- a/src/app/api/auth/codex/device-poll/route.ts +++ b/src/app/api/auth/codex/device-poll/route.ts @@ -9,6 +9,7 @@ import { auditLog } from "@/lib/auth/audit"; import { getClientIp, apiSuccess, apiError } from "@/lib/api-response"; import { annotate } from "@/lib/logging/context"; import { INSIGHTS_CACHE_CLEARED } from "@/lib/insights/cached-text-cipher"; +import { forgetProviderHealth } from "@/lib/ai/provider-health-ledger"; /** * v1.4.7.1: Device-code flow polling endpoint. @@ -88,6 +89,9 @@ export const POST = apiHandler(async (request: NextRequest) => { }); cookieStore.delete("codex_device"); + // A new sign-in: whatever the ledger held against the old one no longer + // describes this credential. + await forgetProviderHealth({ userId: user.id, providerType: "codex" }); await auditLog("codex.oauth.connected", { userId: user.id, diff --git a/src/app/api/dashboard/summary/__tests__/route.test.ts b/src/app/api/dashboard/summary/__tests__/route.test.ts index b74b91c07..c6fd2ca31 100644 --- a/src/app/api/dashboard/summary/__tests__/route.test.ts +++ b/src/app/api/dashboard/summary/__tests__/route.test.ts @@ -78,6 +78,11 @@ vi.mock("@/lib/rollups/measurement-read", () => ({ // covered by `src/lib/analytics/__tests__/mood-series.test.ts`. Pinned to an // empty series by default so the legacy assertions keep their "no mood" // expectation, and overridden per test below. +vi.mock("@/lib/dashboard/summary-reads", async (importOriginal) => ({ + ...(await importOriginal()), + readAllTimeCounts: vi.fn(), +})); + vi.mock("@/lib/analytics/mood-series", () => ({ buildMoodDailySeries: vi.fn(async () => ({ entries: [], @@ -106,6 +111,7 @@ import { getSession } from "@/lib/auth/session"; import { __resetAllCachesForTests } from "@/lib/cache/server-cache"; import { buildMoodDailySeries } from "@/lib/analytics/mood-series"; import { userDayKey } from "@/lib/tz/format"; +import { readAllTimeCounts } from "@/lib/dashboard/summary-reads"; const SESSION_OK = { session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) }, @@ -133,10 +139,10 @@ beforeEach(() => { // between tests to keep cross-test isolation. __resetAllCachesForTests(); vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); - // v1.4.33 — the route now issues a `groupBy` for - // per-type all-time count + most-recent timestamp. Default to an - // empty aggregate so legacy tests keep their "no data" expectations. - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([] as never); + // The per-type all-time counts (`readAllTimeCounts`). Default to none so + // legacy tests keep their "no data" expectations; the newest reading per + // type, which `lastSeenAt` carries, comes from the latest-reading read. + vi.mocked(readAllTimeCounts).mockResolvedValue([] as never); // v1.11.4 — the sleep tile reads the raw per-stage SLEEP_DURATION rows // via `measurement.findMany` to reconstruct the night total. Default to // no rows so legacy tests keep their "no sleep" expectation. @@ -249,22 +255,26 @@ describe("GET /api/dashboard/summary", () => { } }); - it("surfaces lastSeenAt from the historical aggregate even when the 7-day window is empty", async () => { + it("surfaces lastSeenAt from the newest reading even when the 7-day window is empty", async () => { // Power-user path — the user has logged weight for years but - // hasn't touched the app in two weeks. `groupBy` returns the - // all-time count + most-recent timestamp; the 7-day `findMany` - // returns empty. The route should still emit the weight card + // hasn't touched the app in two weeks. The all-time count and the + // newest reading come back; the sparkline is empty. The route should + // still emit the weight card // with the historical `lastSeenAt` so the iOS client renders the // staleness caption. vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); const twoWeeksAgo = new Date(Date.now() - 14 * 86_400_000); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "WEIGHT", - _count: { _all: 312 }, - _max: { measuredAt: twoWeeksAgo }, + count: 312, }, ] as never); + // The newest reading, which `lastSeenAt` carries (first `$queryRaw` of + // the round: the latest-reading read). + vi.mocked(prisma.$queryRaw).mockResolvedValueOnce([ + { type: "WEIGHT", value: 81.2, measured_at: twoWeeksAgo }, + ] as never); const res = await callGet(makeReq()); const body = (await res.json()) as { data: { @@ -281,10 +291,9 @@ describe("GET /api/dashboard/summary", () => { expect(weight, "weight card must be present").toBeDefined(); expect(weight?.allTimeCount).toBe(312); expect(weight?.lastSeenAt).toBe(twoWeeksAgo.toISOString()); - // No 7-day reading → latestValue stays null but updatedAt falls - // through to the historical timestamp so the iOS client can - // build the relative-age caption from a single field. - expect(weight?.latestValue).toBeNull(); + // The tile shows the historical reading (REG-11) and dates it, so the + // iOS client builds the relative-age caption from one field. + expect(weight?.latestValue).toBe(81.2); expect(weight?.updatedAt).toBe(twoWeeksAgo.toISOString()); }); @@ -299,16 +308,14 @@ describe("GET /api/dashboard/summary", () => { vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); const sixtyDaysAgo = new Date(Date.now() - 60 * 86_400_000); const sixtyOneDaysAgo = new Date(Date.now() - 61 * 86_400_000); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "BLOOD_PRESSURE_SYS", - _count: { _all: 14 }, - _max: { measuredAt: sixtyDaysAgo }, + count: 14, }, { type: "BLOOD_PRESSURE_DIA", - _count: { _all: 14 }, - _max: { measuredAt: sixtyDaysAgo }, + count: 14, }, ] as never); // Promise.all order: [latestEver, sparkline, streakDays]. @@ -406,11 +413,10 @@ describe("GET /api/dashboard/summary", () => { // don't accidentally regress the fresh-data case. vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); const twoDaysAgo = new Date(Date.now() - 2 * 86_400_000); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "WEIGHT", - _count: { _all: 200 }, - _max: { measuredAt: twoDaysAgo }, + count: 200, }, ] as never); vi.mocked(prisma.$queryRaw) @@ -457,11 +463,10 @@ describe("GET /api/dashboard/summary", () => { // oldest bucket first. vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); const days = (n: number) => new Date(Date.now() - n * 86_400_000); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "PULSE", - _count: { _all: 42 }, - _max: { measuredAt: days(30) }, + count: 42, }, ] as never); vi.mocked(prisma.$queryRaw) @@ -547,13 +552,15 @@ describe("GET /api/dashboard/summary", () => { // hint). vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); const tenDaysAgo = new Date(Date.now() - 10 * 86_400_000); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "BLOOD_GLUCOSE", - _count: { _all: 4 }, - _max: { measuredAt: tenDaysAgo }, + count: 4, }, ] as never); + vi.mocked(prisma.$queryRaw).mockResolvedValueOnce([ + { type: "BLOOD_GLUCOSE", value: 104, measured_at: tenDaysAgo }, + ] as never); const res = await callGet(makeReq()); const body = (await res.json()) as { data: { @@ -579,11 +586,10 @@ describe("GET /api/dashboard/summary", () => { // hours, NOT surface a single stage. Excludes IN_BED + AWAKE. vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); const wake = new Date("2026-06-04T06:00:00.000Z"); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "SLEEP_DURATION", - _count: { _all: 5 }, - _max: { measuredAt: wake }, + count: 5, }, ] as never); // Raw per-stage rows for the night (the sleep findMany read). @@ -654,11 +660,10 @@ describe("GET /api/dashboard/summary", () => { // total; the additive annotation surfaces the disagreement. vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); const wake = new Date("2026-06-04T06:00:00.000Z"); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "SLEEP_DURATION", - _count: { _all: 4 }, - _max: { measuredAt: wake }, + count: 4, }, ] as never); vi.mocked(prisma.measurement.findMany).mockResolvedValue([ @@ -780,11 +785,10 @@ describe("GET /api/dashboard/summary", () => { // returned shape proves the route hands `sum_value` through // instead of reconstructing from `mean * count`. vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "ACTIVITY_STEPS", - _count: { _all: 7 }, - _max: { measuredAt: new Date() }, + count: 7, }, ] as never); // First call: latest7d. Second call: sparkline. Third call: @@ -824,11 +828,10 @@ describe("GET /api/dashboard/summary", () => { it("falls back to mean * count when the legacy sum_value is null (v1.4.39 W-SUM)", async () => { vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); - vi.mocked(prisma.measurement.groupBy).mockResolvedValue([ + vi.mocked(readAllTimeCounts).mockResolvedValue([ { type: "ACTIVITY_STEPS", - _count: { _all: 7 }, - _max: { measuredAt: new Date() }, + count: 7, }, ] as never); vi.mocked(prisma.$queryRaw) diff --git a/src/app/api/dashboard/summary/route.ts b/src/app/api/dashboard/summary/route.ts index 41a60ae4a..4526bf946 100644 --- a/src/app/api/dashboard/summary/route.ts +++ b/src/app/api/dashboard/summary/route.ts @@ -79,6 +79,7 @@ import { buildMedsTodayBlock } from "@/lib/dashboard/meds-today"; import { computeStreak } from "@/lib/dashboard/streak"; import { readActivityDayKeys, + readAllTimeCounts, readLatestEver, readSparkBuckets, } from "@/lib/dashboard/summary-reads"; @@ -468,7 +469,7 @@ async function buildDashboardSummary( // - sparkBuckets: ≤ SPARK_DAYS × N metric types (typically <80, // REG-11: taken via ROW_NUMBER window, no calendar // filter, so a 60-day-old metric still paints a chart) - // - allTimeAggregate: ≤ N metric types (unchanged) + // - allTimeAggregate: ≤ N metric types, summed from the DAY rollups // - medsToday: today-window tally via the shared projector // (≤ daily intake schedule count, unchanged semantics) // - streakActivity: ≤ daily intake count × 365 (unchanged) @@ -492,13 +493,10 @@ async function buildDashboardSummary( time("sparkline", () => readSparkBuckets(prisma, userId, measurementTypes, SPARK_DAYS), ), + // Counted from the DAY rollups where they cover the days, not by one + // index entry per reading (see `readAllTimeCounts`). time("allTime", () => - prisma.measurement.groupBy({ - by: ["type"], - where: { userId, type: { in: measurementTypes }, deletedAt: null }, - _count: { _all: true }, - _max: { measuredAt: true }, - }), + readAllTimeCounts(prisma, userId, measurementTypes, now), ), // v1.4.39 — the dashboard compliance tile pulled // `MedicationIntakeEvent` rows for the today-window only. Daily @@ -583,17 +581,20 @@ async function buildDashboardSummary( // Per-type metadata lookup — typed Map so a metric with no readings // at all falls through `metaForType` to the `{ allTimeCount: 0, - // lastSeenAt: null }` default. The aggregate row's `_count._all` is - // always `number` per Prisma's runtime contract; defending against - // undefined keeps the helper's narrow signature stable. + // lastSeenAt: null }` default. const allTimeByType = new Map< MeasurementType, { allTimeCount: number; lastSeenAt: Date | null } >(); + // The newest reading is the one `latestEver` already read per type; its + // instant is the metric's last-seen time. + const lastSeenByType = new Map( + latestEver.map((row) => [row.type, row.measured_at]), + ); for (const row of allTimeAggregate) { allTimeByType.set(row.type, { - allTimeCount: row._count?._all ?? 0, - lastSeenAt: row._max?.measuredAt ?? null, + allTimeCount: row.count, + lastSeenAt: lastSeenByType.get(row.type) ?? null, }); } function metaForType(type: MeasurementType): { diff --git a/src/app/api/import/__tests__/route.test.ts b/src/app/api/import/__tests__/route.test.ts index 096b72f8c..ef9e668fe 100644 --- a/src/app/api/import/__tests__/route.test.ts +++ b/src/app/api/import/__tests__/route.test.ts @@ -54,9 +54,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { const actual = await vi.importActual< typeof import("@/lib/rollups/measurement-rollups") >("@/lib/rollups/measurement-rollups"); + const perKey = vi.fn().mockResolvedValue(undefined); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForMeasurement: perKey, + // The keyed refold, observed as the per-key recomputes it stands for. + recomputeBucketsForKeys: vi.fn( + async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await perKey(userId, k.type, k.measuredAt); + }, + ), }; }); diff --git a/src/app/api/import/csv/__tests__/route.test.ts b/src/app/api/import/csv/__tests__/route.test.ts index 3ca886bea..a6d8dcaa9 100644 --- a/src/app/api/import/csv/__tests__/route.test.ts +++ b/src/app/api/import/csv/__tests__/route.test.ts @@ -58,9 +58,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { const actual = await vi.importActual< typeof import("@/lib/rollups/measurement-rollups") >("@/lib/rollups/measurement-rollups"); + const perKey = vi.fn().mockResolvedValue(undefined); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForMeasurement: perKey, + // The keyed refold, observed as the per-key recomputes it stands for. + recomputeBucketsForKeys: vi.fn( + async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await perKey(userId, k.type, k.measuredAt); + }, + ), }; }); diff --git a/src/app/api/integrations/status/__tests__/route.test.ts b/src/app/api/integrations/status/__tests__/route.test.ts index 6b8d0e68d..43596e00d 100644 --- a/src/app/api/integrations/status/__tests__/route.test.ts +++ b/src/app/api/integrations/status/__tests__/route.test.ts @@ -485,3 +485,26 @@ describe("/api/integrations/status — the unified connection health", () => { } }); }); + +describe("/api/integrations/status — one round of reads", () => { + it("asks for the connection health and the freshness without waiting on the provider rows", async () => { + userFind.mockResolvedValue({}); + // Hold the user row back: under the old shape nothing else was asked + // until it (and the rest of the first round) had answered. + let release: (row: unknown) => void = () => {}; + userFind.mockImplementationOnce( + () => new Promise((resolve) => (release = resolve)), + ); + const { loadConnectionHealth } = + await import("@/lib/integrations/connection-health-load"); + + const pending = fetchEnvelope(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(loadConnectionHealth).toHaveBeenCalledTimes(1); + expect(listLiveMeasurementTypes).toHaveBeenCalled(); + release({}); + const envelope = await pending; + expect(envelope.integrations.length).toBeGreaterThan(0); + }); +}); diff --git a/src/app/api/integrations/status/route.ts b/src/app/api/integrations/status/route.ts index 1c0d439c3..73396103d 100644 --- a/src/app/api/integrations/status/route.ts +++ b/src/app/api/integrations/status/route.ts @@ -64,124 +64,138 @@ export const GET = apiHandler(async () => { const { user } = await requireAuth(); annotate({ action: { name: "integrations.status" } }); + // One round: the per-provider rows, the unified connection health and the + // per-metric freshness do not depend on each other. Awaited one after + // another, a cold first visit paid the three latencies end to end. + let metricFreshnessDegraded = false; const [ - withingsStatus, - whoopStatus, - fitbitStatus, - googleHealthStatus, - polarStatus, - ouraStatus, - stravaStatus, - nightscoutStatus, - dawarichStatus, - dbUser, - withingsConn, - whoopConn, - fitbitConn, - googleHealthConn, - // v1.17.1 — `available` reports whether usable OAuth credentials resolve - // (per-user BYO first, then the shared env app), mirroring the per-card - // /api//status the consolidated envelope now replaces. - polarAvailable, - ouraAvailable, - stravaAvailable, + [ + withingsStatus, + whoopStatus, + fitbitStatus, + googleHealthStatus, + polarStatus, + ouraStatus, + stravaStatus, + nightscoutStatus, + dawarichStatus, + dbUser, + withingsConn, + whoopConn, + fitbitConn, + googleHealthConn, + // v1.17.1 — `available` reports whether usable OAuth credentials resolve + // (per-user BYO first, then the shared env app), mirroring the per-card + // /api//status the consolidated envelope now replaces. + polarAvailable, + ouraAvailable, + stravaAvailable, + ], + health, + metricFreshness, ] = await Promise.all([ - getIntegrationStatus(user.id, "withings"), - getIntegrationStatus(user.id, "whoop"), - getIntegrationStatus(user.id, "fitbit"), - getIntegrationStatus(user.id, "google-health"), - getIntegrationStatus(user.id, "polar"), - getIntegrationStatus(user.id, "oura"), - getIntegrationStatus(user.id, "strava"), - getIntegrationStatus(user.id, "nightscout"), - getIntegrationStatus(user.id, "dawarich"), - prisma.user.findUnique({ - where: { id: user.id }, - select: { - withingsClientIdEncrypted: true, - withingsClientSecretEncrypted: true, - whoopClientIdEncrypted: true, - whoopClientSecretEncrypted: true, - fitbitClientIdEncrypted: true, - fitbitClientSecretEncrypted: true, - googleHealthClientIdEncrypted: true, - googleHealthClientSecretEncrypted: true, - polarAccessTokenEncrypted: true, - polarClientIdEncrypted: true, - polarClientSecretEncrypted: true, - ouraAccessTokenEncrypted: true, - ouraClientIdEncrypted: true, - ouraClientSecretEncrypted: true, - ouraScopeDenied: true, - stravaAccessTokenEncrypted: true, - stravaClientIdEncrypted: true, - stravaClientSecretEncrypted: true, - // Nightscout folds onto the envelope; the card reads these three off - // it instead of its own /api/nightscout/status round-trip. - nightscoutUrlEncrypted: true, - nightscoutTokenEncrypted: true, - nightscoutAllowPrivateHost: true, - dawarichUrlEncrypted: true, - }, + Promise.all([ + getIntegrationStatus(user.id, "withings"), + getIntegrationStatus(user.id, "whoop"), + getIntegrationStatus(user.id, "fitbit"), + getIntegrationStatus(user.id, "google-health"), + getIntegrationStatus(user.id, "polar"), + getIntegrationStatus(user.id, "oura"), + getIntegrationStatus(user.id, "strava"), + getIntegrationStatus(user.id, "nightscout"), + getIntegrationStatus(user.id, "dawarich"), + prisma.user.findUnique({ + where: { id: user.id }, + select: { + withingsClientIdEncrypted: true, + withingsClientSecretEncrypted: true, + whoopClientIdEncrypted: true, + whoopClientSecretEncrypted: true, + fitbitClientIdEncrypted: true, + fitbitClientSecretEncrypted: true, + googleHealthClientIdEncrypted: true, + googleHealthClientSecretEncrypted: true, + polarAccessTokenEncrypted: true, + polarClientIdEncrypted: true, + polarClientSecretEncrypted: true, + ouraAccessTokenEncrypted: true, + ouraClientIdEncrypted: true, + ouraClientSecretEncrypted: true, + ouraScopeDenied: true, + stravaAccessTokenEncrypted: true, + stravaClientIdEncrypted: true, + stravaClientSecretEncrypted: true, + // Nightscout folds onto the envelope; the card reads these three off + // it instead of its own /api/nightscout/status round-trip. + nightscoutUrlEncrypted: true, + nightscoutTokenEncrypted: true, + nightscoutAllowPrivateHost: true, + dawarichUrlEncrypted: true, + }, + }), + prisma.withingsConnection.findUnique({ + where: { userId: user.id }, + select: { + tokenExpiresAt: true, + lastSyncedAt: true, + createdAt: true, + scope: true, + apiPlanState: true, + }, + }), + prisma.whoopConnection.findUnique({ + where: { userId: user.id }, + select: { + tokenExpiresAt: true, + whoopUserId: true, + lastSyncedAt: true, + createdAt: true, + backfillCompletedAt: true, + }, + }), + prisma.fitbitConnection.findUnique({ + where: { userId: user.id }, + select: { + tokenExpiresAt: true, + lastSyncedAt: true, + createdAt: true, + backfillCompletedAt: true, + }, + }), + prisma.googleHealthConnection.findUnique({ + where: { userId: user.id }, + select: { + tokenExpiresAt: true, + lastSyncedAt: true, + createdAt: true, + backfillCompletedAt: true, + backfillProgress: true, + needsReauth: true, + }, + }), + getPolarClientCredentials(user.id).then((c) => !!c), + getOuraClientCredentials(user.id).then((c) => !!c), + getStravaClientCredentials(user.id).then((c) => !!c), + ]), + // The unified per-connection state. Resolved by the same loader the + // per-provider routes and the integration watchdog use, so the envelope + // and every other surface say the same thing about a connection. + loadConnectionHealth(user.id, [ + "withings", + "whoop", + "fitbit", + "google-health", + "polar", + "oura", + "strava", + "nightscout", + "dawarich", + ]), + // See `metricFreshnessDegraded` below. + getSourceMetricFreshness(user.id).catch(() => { + metricFreshnessDegraded = true; + return {} as Partial>; }), - prisma.withingsConnection.findUnique({ - where: { userId: user.id }, - select: { - tokenExpiresAt: true, - lastSyncedAt: true, - createdAt: true, - scope: true, - apiPlanState: true, - }, - }), - prisma.whoopConnection.findUnique({ - where: { userId: user.id }, - select: { - tokenExpiresAt: true, - whoopUserId: true, - lastSyncedAt: true, - createdAt: true, - backfillCompletedAt: true, - }, - }), - prisma.fitbitConnection.findUnique({ - where: { userId: user.id }, - select: { - tokenExpiresAt: true, - lastSyncedAt: true, - createdAt: true, - backfillCompletedAt: true, - }, - }), - prisma.googleHealthConnection.findUnique({ - where: { userId: user.id }, - select: { - tokenExpiresAt: true, - lastSyncedAt: true, - createdAt: true, - backfillCompletedAt: true, - backfillProgress: true, - needsReauth: true, - }, - }), - getPolarClientCredentials(user.id).then((c) => !!c), - getOuraClientCredentials(user.id).then((c) => !!c), - getStravaClientCredentials(user.id).then((c) => !!c), - ]); - - // The unified per-connection state. Resolved by the same loader the - // per-provider routes and the integration watchdog use, so the envelope and - // every other surface say the same thing about a connection. - const health = await loadConnectionHealth(user.id, [ - "withings", - "whoop", - "fitbit", - "google-health", - "polar", - "oura", - "strava", - "nightscout", - "dawarich", ]); // Per-metric-type last-value timestamps so the card can show that a @@ -196,12 +210,7 @@ export const GET = apiHandler(async () => { // the wire, and the honesty signal quietly became the thing it was built to // prevent. `metricFreshnessDegraded` separates them. One flag rather than a // nullable array per entry, because this is ONE query for all eight providers - // — it fails for all of them or for none. - let metricFreshnessDegraded = false; - const metricFreshness = await getSourceMetricFreshness(user.id).catch(() => { - metricFreshnessDegraded = true; - return {} as Partial>; - }); + // — it fails for all of them or for none. (Read in the round above.) const now = Date.now(); const whoopConnected = !!whoopConn?.whoopUserId; diff --git a/src/app/api/measurements/bulk-delete/__tests__/route.test.ts b/src/app/api/measurements/bulk-delete/__tests__/route.test.ts index 3a3438a12..b41d53ccb 100644 --- a/src/app/api/measurements/bulk-delete/__tests__/route.test.ts +++ b/src/app/api/measurements/bulk-delete/__tests__/route.test.ts @@ -47,10 +47,20 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { const actual = await vi.importActual< typeof import("@/lib/rollups/measurement-rollups") >("@/lib/rollups/measurement-rollups"); + const perKey = vi.fn().mockResolvedValue(undefined); return { // keep the real collapse so the test exercises the actual fan-out logic collapseToTypeDayKeys: actual.collapseToTypeDayKeys, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForMeasurement: perKey, + // The keyed refold, observed as the per-key recomputes it stands for. + recomputeBucketsForKeys: vi.fn( + async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await perKey(userId, k.type, k.measuredAt); + }, + ), }; }); vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); diff --git a/src/app/api/measurements/restore/__tests__/route.test.ts b/src/app/api/measurements/restore/__tests__/route.test.ts index 8ef10c8c3..0fa7e8427 100644 --- a/src/app/api/measurements/restore/__tests__/route.test.ts +++ b/src/app/api/measurements/restore/__tests__/route.test.ts @@ -47,10 +47,20 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { const actual = await vi.importActual< typeof import("@/lib/rollups/measurement-rollups") >("@/lib/rollups/measurement-rollups"); + const perKey = vi.fn().mockResolvedValue(undefined); return { // keep the real collapse so the test exercises the actual fan-out logic collapseToTypeDayKeys: actual.collapseToTypeDayKeys, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForMeasurement: perKey, + // The keyed refold, observed as the per-key recomputes it stands for. + recomputeBucketsForKeys: vi.fn( + async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await perKey(userId, k.type, k.measuredAt); + }, + ), }; }); vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); diff --git a/src/lib/ai/provider-health-ledger.ts b/src/lib/ai/provider-health-ledger.ts index 1ffbb461c..b6e25a4b6 100644 --- a/src/lib/ai/provider-health-ledger.ts +++ b/src/lib/ai/provider-health-ledger.ts @@ -266,6 +266,33 @@ export const postgresProviderHealthLedger: ProviderHealthLedger = { }, }; +/** + * Drop the recorded outcomes of a credential that has just been replaced. + * + * The ledger describes a credential, not a provider slot: once the person + * (or, for `admin-codex`, the operator) signs in again, the `auth_failed` + * row of the refused credential would otherwise keep the fresh one at the + * tail of the chain and reported as expired for the rest of its cooldown. + * Without `userId` the rows of every account are dropped — the operator's + * central connection is one credential shared by all of them. Best effort, + * like every ledger write. + */ +export async function forgetProviderHealth(target: { + providerType: ProviderChainType; + userId?: string; +}): Promise { + try { + await prisma.providerHealth.deleteMany({ + where: { + providerType: target.providerType, + ...(target.userId ? { userId: target.userId } : {}), + }, + }); + } catch { + // A stale row only deprioritises; it never blocks. + } +} + /** * Providers whose credential is currently known-bad (auth-class failure * inside the cooldown window) for a user. Drives the proactive diff --git a/src/lib/ai/provider.ts b/src/lib/ai/provider.ts index 51f99b5c6..e05758bed 100644 --- a/src/lib/ai/provider.ts +++ b/src/lib/ai/provider.ts @@ -30,6 +30,7 @@ import { type ProviderWorkAuthority, } from "@/lib/sharing/provider-work-authority"; import { memoizePerRequest } from "@/lib/request-cache"; +import { emitSignal } from "@/lib/logging/signal"; import type { AiModality } from "@/lib/ai/capabilities/types"; import type { ProviderEntryPresence, @@ -42,6 +43,93 @@ import { const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; // 5 minutes +/** + * Whether a failed Codex refresh proves the sign-in dead: the issuer refused + * the grant itself (`CodexRefreshError.credentialExpired`), as opposed to a + * 429, a 5xx or a network error, which say nothing about the credential. + * Read by shape rather than `instanceof` so a mocked `codex-oauth` module + * classifies the same way. + */ +function isDeadCodexSignIn(error: unknown): boolean { + return ( + (error as { credentialExpired?: unknown } | null)?.credentialExpired === + true + ); +} + +function oauthErrorOf(error: unknown): string | null { + const code = (error as { oauthError?: unknown } | null)?.oauthError; + return typeof code === "string" ? code : null; +} + +/** + * Park a Codex connection whose sign-in the issuer has refused for good. + * + * Retrying a dead refresh token cannot succeed; until this existed every + * generation of every card of every run asked the issuer again, failed with + * the same 401, and logged it. Parking flips the connection to the existing + * `expired` state — the one Settings already renders as "reconnect ChatGPT" — + * so the chain, the presence probe and with it every AI capability stop + * counting it, and nothing is sent until the person signs in again (the + * device-code flow writes `connected`). + * + * Compare-and-set on the refresh token that was refused: a reconnect that + * landed in between wrote a different one and is left alone, and of several + * workers failing at once exactly one flips the row, so the transition is + * signalled once and never again per run. + */ +async function parkUserCodex( + userId: string, + refusedRefreshEncrypted: string, + error: unknown, +): Promise { + try { + const parked = await prisma.user.updateMany({ + where: { + id: userId, + codexConnectionStatus: "connected", + codexRefreshTokenEncrypted: refusedRefreshEncrypted, + }, + data: { codexConnectionStatus: "expired" }, + }); + if (parked.count === 1) { + emitSignal({ + action: "ai.credential.expired", + level: "warn", + meta: { provider: "codex", oauth_error: oauthErrorOf(error) }, + }); + } + } catch { + // Best effort: an unparked connection is retried exactly as before. + } +} + +/** The operator's central Codex, parked the same way. */ +async function parkCentralCodex( + refusedRefreshEncrypted: string, + error: unknown, +): Promise { + try { + const parked = await prisma.appSettings.updateMany({ + where: { + id: "singleton", + adminCodexConnectionStatus: "connected", + adminCodexRefreshTokenEncrypted: refusedRefreshEncrypted, + }, + data: { adminCodexConnectionStatus: "expired" }, + }); + if (parked.count === 1) { + emitSignal({ + action: "ai.credential.expired", + level: "warn", + meta: { provider: "admin-codex", oauth_error: oauthErrorOf(error) }, + }); + } + } catch { + // Best effort, as above. + } +} + class NoProvider implements AIProvider { readonly type = "none" as const; @@ -308,9 +396,15 @@ async function resolveCodexProvider( codexTokenExpiresAt: fresh.expiresAt, }, }); - } catch { - // Fall through — CodexClient will trigger an on-401 refresh - // and persist there. + } catch (error) { + // A refused grant parks the connection and drops it from this chain: + // calling Codex with the expired access token could only 401 into a + // second refusal. Anything else falls through — CodexClient will + // trigger an on-401 refresh and persist there. + if (isDeadCodexSignIn(error)) { + await parkUserCodex(userId, user.codexRefreshTokenEncrypted, error); + return null; + } } } @@ -336,9 +430,27 @@ async function resolveCodexProvider( refreshEncrypted: freshUser.codexRefreshTokenEncrypted, }); if (!decoded) { - throw codexReconnectRequired("token storage corrupt"); + const error = codexReconnectRequired("token storage corrupt"); + await parkUserCodex( + userId, + freshUser.codexRefreshTokenEncrypted, + error, + ); + throw error; + } + let fresh; + try { + fresh = await refreshDeviceTokens(decoded.refreshToken); + } catch (error) { + if (isDeadCodexSignIn(error)) { + await parkUserCodex( + userId, + freshUser.codexRefreshTokenEncrypted, + error, + ); + } + throw error; } - const fresh = await refreshDeviceTokens(decoded.refreshToken); const enc = encryptCodexCreds(fresh); await prisma.user.update({ where: { id: userId }, @@ -421,9 +533,14 @@ async function resolveAdminCodexProvider(): Promise { adminCodexTokenExpiresAt: enc.expiresAt, }, }); - } catch { - // Fall through — CodexClient triggers an on-401 refresh and persists via - // the callback below. + } catch (error) { + // A refused grant parks the central connection (see `parkUserCodex`); + // anything else falls through — CodexClient triggers an on-401 refresh + // and persists via the callback below. + if (isDeadCodexSignIn(error)) { + await parkCentralCodex(settings.adminCodexRefreshTokenEncrypted, error); + return null; + } } } @@ -454,9 +571,19 @@ async function resolveAdminCodexProvider(): Promise { expiresAt: fresh.adminCodexTokenExpiresAt, }); if (!decoded) { - throw codexReconnectRequired("central token storage corrupt"); + const error = codexReconnectRequired("central token storage corrupt"); + await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error); + throw error; + } + let refreshed; + try { + refreshed = await refreshDeviceTokens(decoded.refreshToken); + } catch (error) { + if (isDeadCodexSignIn(error)) { + await parkCentralCodex(fresh.adminCodexRefreshTokenEncrypted, error); + } + throw error; } - const refreshed = await refreshDeviceTokens(decoded.refreshToken); const enc = encryptAdminCodexCreds(refreshed); await prisma.appSettings.update({ where: { id: "singleton" }, diff --git a/src/lib/dashboard/summary-reads.ts b/src/lib/dashboard/summary-reads.ts index 1a0322d63..8f8550160 100644 --- a/src/lib/dashboard/summary-reads.ts +++ b/src/lib/dashboard/summary-reads.ts @@ -17,6 +17,7 @@ */ import { Prisma, type PrismaClient } from "@/generated/prisma/client"; import type { MeasurementType } from "@/generated/prisma/client"; +import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups"; /** v1.4.38 — the single most recent reading per measurement * type, irrespective of age. One row per metric the user has ever @@ -166,6 +167,82 @@ export function readSparkBuckets( `; } +/** One metric's all-time reading count, for a type that has any. */ +export interface AllTimeCountRow { + type: MeasurementType; + count: number; +} + +/** + * The first UTC day the DAY rollups are trusted to hold in full: one day + * inside the fold window. Every fold that ever ran for the account began at + * or before it (its window was anchored on an earlier `now`), and the write + * hooks keep every later day current. + */ +export function allTimeRollupBoundary(now: Date): Date { + const DAY_MS = 24 * 60 * 60 * 1000; + const start = now.getTime() - ROLLUP_FOLD_WINDOW_MS; + return new Date(Math.floor(start / DAY_MS) * DAY_MS + DAY_MS); +} + +/** + * Total live readings per measurement type, all time. + * + * Counting them on `measurements` reads one index entry per reading: on an + * account with a watch that is millions of entries per request, two to four + * seconds whenever they are not in memory, which is the dashboard's slowest + * read by far. The DAY rollups already carry a count per (type, day, source) + * — every live reading of the day, every source — so the sum of a type's DAY + * buckets is its count over the days they cover. Readings before the fold + * window, which the fold never buckets, are counted live; that range is a + * small, bounded index slice. A type with no DAY bucket at all (not folded + * yet) is counted live in full, exactly as before. Types with no readings + * are absent. + */ +export async function readAllTimeCounts( + db: PrismaClient, + userId: string, + measurementTypes: readonly MeasurementType[], + now: Date = new Date(), +): Promise { + const boundary = allTimeRollupBoundary(now); + const rows = await db.$queryRaw< + Array<{ type: MeasurementType; count: number }> + >` + SELECT wanted."type" AS type, + (CASE + WHEN EXISTS ( + SELECT 1 FROM measurement_rollups r + WHERE r."user_id" = ${userId} + AND r."type" = wanted."type" + AND r."granularity" = 'DAY' + ) + THEN COALESCE(( + SELECT SUM(r."count") FROM measurement_rollups r + WHERE r."user_id" = ${userId} + AND r."type" = wanted."type" + AND r."granularity" = 'DAY' + AND r."bucket_start" >= ${boundary} + ), 0) + ( + SELECT COUNT(*) FROM measurements m + WHERE m."user_id" = ${userId} + AND m."type" = wanted."type" + AND m."deleted_at" IS NULL + AND m."measured_at" < ${boundary} + ) + ELSE ( + SELECT COUNT(*) FROM measurements m + WHERE m."user_id" = ${userId} + AND m."type" = wanted."type" + AND m."deleted_at" IS NULL + ) + END)::int AS count + FROM unnest(${[...measurementTypes]}::"measurement_type"[]) AS wanted("type") + ORDER BY wanted."type" + `; + return rows.filter((row) => row.count > 0); +} + /** * The local days in `[since, now]` that hold at least one live reading, as * `YYYY-MM-DD` keys in `timeZone`: the measurement half of the streak. diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index c636b3071..e52c656c9 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/src/lib/fitbit/sync-core.ts @@ -18,7 +18,7 @@ import { getEvent } from "@/lib/logging/context"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { refreshAccessToken } from "./client"; import { getUserFitbitCredentials } from "./credentials"; @@ -894,9 +894,7 @@ export async function upsertFitbitMeasurements( try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); invalidateStatusInsightsForTypes( userId, keys.map((k) => k.type), diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index 542755bc8..4741df368 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -41,7 +41,7 @@ import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { @@ -1587,9 +1587,7 @@ export async function upsertGoogleHealthMeasurements( try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); invalidateStatusInsightsForTypes( userId, keys.map((k) => k.type), diff --git a/src/lib/jobs/insight-pregenerate.ts b/src/lib/jobs/insight-pregenerate.ts index c67b1c929..509551b51 100644 --- a/src/lib/jobs/insight-pregenerate.ts +++ b/src/lib/jobs/insight-pregenerate.ts @@ -888,7 +888,8 @@ export async function runInsightPregenerate( } else if (generated.credentialExpired) { // v1.42 — a dead sign-in (a Codex refresh token the issuer // revoked) does not heal in 45 minutes either; the person has - // to reconnect, and the ledger already tells them so. + // to reconnect. The refusal parked the connection as expired, + // which Settings shows, and the next run no longer counts it. settle("auth_failed", generated.reason); } else { settle("failed", generated.reason); diff --git a/src/lib/nightscout/__tests__/sync.test.ts b/src/lib/nightscout/__tests__/sync.test.ts index e43560469..18f3c8a9b 100644 --- a/src/lib/nightscout/__tests__/sync.test.ts +++ b/src/lib/nightscout/__tests__/sync.test.ts @@ -43,7 +43,12 @@ vi.mock("@/lib/integrations/status", async (importOriginal) => ({ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) => rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), - recomputeBucketsForMeasurement: recomputeMock, + recomputeBucketsForKeys: async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await recomputeMock(userId, k.type, k.measuredAt); + }, })); vi.mock("@/lib/insights/comprehensive-generate", () => ({ diff --git a/src/lib/nightscout/sync.ts b/src/lib/nightscout/sync.ts index c1c0edc92..fa3998a0b 100644 --- a/src/lib/nightscout/sync.ts +++ b/src/lib/nightscout/sync.ts @@ -34,7 +34,7 @@ import type { IntegrationClassification } from "@/lib/integrations/http-status-c import type { SyncWriteResult } from "@/lib/outcome/written-outcome"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { emitInsertedMeasurementArrivals, @@ -327,9 +327,7 @@ export async function upsertNightscoutEntries( try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); invalidateStatusInsightsForTypes( userId, keys.map((k) => k.type), diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index 299f6e892..1f2d22728 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/src/lib/oura/__tests__/sync.test.ts @@ -98,7 +98,12 @@ vi.mock("@/lib/integrations/status", async (importOriginal) => ({ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) => rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), - recomputeBucketsForMeasurement: recomputeMock, + recomputeBucketsForKeys: async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await recomputeMock(userId, k.type, k.measuredAt); + }, })); vi.mock("@/lib/insights/comprehensive-generate", () => ({ diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index d401fa2b6..0a5c9b384 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -45,7 +45,7 @@ import { } from "@/lib/integrations/status"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { emitInsertedMeasurementArrivals, @@ -805,9 +805,7 @@ export async function upsertOuraMeasurements( ); try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); invalidateStatusInsightsForTypes( userId, keys.map((k) => k.type), diff --git a/src/lib/polar/sync.ts b/src/lib/polar/sync.ts index 29147bf24..fede4d187 100644 --- a/src/lib/polar/sync.ts +++ b/src/lib/polar/sync.ts @@ -48,7 +48,7 @@ import { } from "@/lib/integrations/status"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; @@ -398,9 +398,7 @@ export async function upsertPolarMeasurements( ); try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); invalidateStatusInsightsForTypes( userId, keys.map((k) => k.type), diff --git a/src/lib/rollups/__tests__/after-measurement-mutation.test.ts b/src/lib/rollups/__tests__/after-measurement-mutation.test.ts index 681356457..cf80d3cb6 100644 --- a/src/lib/rollups/__tests__/after-measurement-mutation.test.ts +++ b/src/lib/rollups/__tests__/after-measurement-mutation.test.ts @@ -4,7 +4,7 @@ * invalidation). * * Watched red: with the recompute loop removed from the helper the - * first test fails naming the missing `recomputeBucketsForMeasurement` + * first test fails naming the missing `recomputeBucketsForKeys` * call; with the `.catch` isolation removed the failure-isolation test * throws. Both verified red against a deliberately broken helper before * the real implementation landed. @@ -27,7 +27,12 @@ vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => { await importOriginal(); return { ...actual, - recomputeBucketsForMeasurement: mocks.recompute, + recomputeBucketsForKeys: async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await mocks.recompute(userId, k.type, k.measuredAt); + }, }; }); diff --git a/src/lib/rollups/after-measurement-mutation.ts b/src/lib/rollups/after-measurement-mutation.ts index f12c17160..aa501d1d6 100644 --- a/src/lib/rollups/after-measurement-mutation.ts +++ b/src/lib/rollups/after-measurement-mutation.ts @@ -8,7 +8,7 @@ * * 1. the `(type, day)` rollup buckets the mutation dirtied — recomputed * inline (DAY) + enqueued (WEEK/MONTH/YEAR) via - * `recomputeBucketsForMeasurement`. The recompute handles deletions + * `recomputeBucketsForKeys`. The recompute handles deletions * inherently: a day left empty drops its rollup row. * 2. the cached per-metric status assessments — re-warmed via * `invalidateStatusInsightsForTypes` (fire-and-forget). @@ -20,7 +20,7 @@ */ import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes, @@ -50,9 +50,7 @@ export async function afterMeasurementMutation( // 500-row batch fires at most one DAY recompute per touched bucket. const keys = collapseToTypeDayKeys(rows); try { - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); } catch (err) { console.warn(`[${label}] rollup recompute failed`, err); } diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index cf33a2c6d..bfaf64b92 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -358,6 +358,143 @@ export async function recomputeBucketsForMeasurement( ); } +/** Keys one set-based DAY refold carries; bounds the statement's arrays. */ +const KEYED_REFOLD_CHUNK = 2_000; + +/** + * The set-based sibling of {@link recomputeBucketsForMeasurement}, for a write + * that touched many `(type, day)` buckets at once (a sync, an import, a batch). + * + * The per-key hook costs an aggregate, a transaction and three queue sends per + * bucket, so a sync that re-read two years of history paid that seven thousand + * times over. Here the DAY tier is refolded for exactly the touched buckets in + * one aggregate and one transaction per chunk, and the coarser tiers are + * enqueued once per contiguous run of buckets rather than once per day. The + * result is the one the per-key loop produced: every touched day re-derived + * from its live rows, a day left empty dropped, untouched days left alone. + */ +export async function recomputeBucketsForKeys( + userId: string, + rows: Array<{ type: MeasurementType; measuredAt: Date }>, +): Promise { + const keys = collapseToTypeDayKeys(rows); + for (let i = 0; i < keys.length; i += KEYED_REFOLD_CHUNK) { + await refoldDayBuckets(userId, keys.slice(i, i + KEYED_REFOLD_CHUNK)); + } + for (const granularity of ASYNC_GRANULARITIES) { + for (const run of bucketRuns(keys, granularity)) { + await enqueueRollupRecompute({ userId, granularity, ...run }); + } + } +} + +async function refoldDayBuckets( + userId: string, + keys: Array<{ type: MeasurementType; measuredAt: Date }>, +): Promise { + if (keys.length === 0) return; + const types = [...new Set(keys.map((k) => k.type))]; + const days = [...new Set(keys.map((k) => k.measuredAt.getTime()))].sort( + (a, b) => a - b, + ); + const wanted = new Set( + keys.map((k) => `${k.type}|${k.measuredAt.getTime()}`), + ); + const aggregate = await runRollupAggregate({ + userId, + types, + granularity: "DAY", + from: new Date(days[0]!), + to: new Date(days[days.length - 1]! + 24 * 60 * 60 * 1000), + days: days.map((d) => new Date(d)), + }); + // The day filter is per day, not per (type, day): another touched type's + // day comes back too. Its aggregate would be right, but it was not touched, + // so it is not rewritten. + const ordered = aggregate + .filter((r) => + wanted.has(`${r.type}|${new Date(r.bucket_start).getTime()}`), + ) + .sort((a, b) => + a.type !== b.type + ? a.type < b.type + ? -1 + : 1 + : a.source !== b.source + ? a.source < b.source + ? -1 + : 1 + : a.bucket_start.getTime() - b.bucket_start.getTime(), + ); + // Delete exactly the touched buckets (every source, so a source that left a + // day does not strand its row, and a day that emptied out is dropped), then + // upsert the fresh aggregate, in one transaction like `persistRollupRows`. + await prisma.$transaction( + async (tx) => { + await tx.$executeRawUnsafe( + `DELETE FROM measurement_rollups + WHERE user_id = $1 + AND granularity = 'DAY'::"measurement_rollup_granularity" + AND (type, bucket_start) IN ( + SELECT k.t::"measurement_type", k.b + FROM unnest($2::text[], $3::timestamptz[]) AS k(t, b) + )`, + userId, + keys.map((k) => k.type), + keys.map((k) => k.measuredAt.toISOString()), + ); + for (let i = 0; i < ordered.length; i += 500) { + const { sql, params } = buildRollupUpsert( + userId, + "DAY", + ordered.slice(i, i + 500), + ); + await tx.$executeRawUnsafe(sql, ...params); + } + }, + { maxWait: 10_000, timeout: 120_000 }, + ); +} + +/** + * The coarse buckets the touched days fall in, merged per type into runs of + * adjacent buckets: one recompute job per run, each covering whole buckets. + */ +function bucketRuns( + keys: Array<{ type: MeasurementType; measuredAt: Date }>, + granularity: RollupGranularity, +): Array<{ type: MeasurementType; from: Date; to: Date }> { + const spans = new Map< + string, + { type: MeasurementType; from: Date; to: Date } + >(); + for (const k of keys) { + const span = bucketSpan(k.measuredAt, granularity); + spans.set(`${k.type}|${span.from.getTime()}`, { type: k.type, ...span }); + } + const sorted = [...spans.values()].sort((a, b) => + a.type !== b.type + ? a.type < b.type + ? -1 + : 1 + : a.from.getTime() - b.from.getTime(), + ); + const runs: Array<{ type: MeasurementType; from: Date; to: Date }> = []; + for (const span of sorted) { + const last = runs[runs.length - 1]; + if ( + last && + last.type === span.type && + last.to.getTime() === span.from.getTime() + ) { + last.to = span.to; + } else { + runs.push({ ...span }); + } + } + return runs; +} + /** * Best-effort pg-boss enqueue. Silent no-op when no boss is attached * (route is running in a context without the worker — typically the @@ -399,7 +536,16 @@ export async function enqueueRollupRecompute(input: { // is still QUEUED — safe here precisely because the handler re-reads the // bucket's live rows when it starts, so the queued job still sees writes // that land after it was enqueued. - singletonKey: `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}`, + // + // A job covering a run of several buckets (`recomputeBucketsForKeys`) + // carries its end in the key too: under the bare start it would collapse + // into a queued single-bucket job and the rest of the run would be lost. + singletonKey: + `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}` + + (bucketSpan(input.from, input.granularity).to.getTime() === + input.to.getTime() + ? "" + : `|${input.to.toISOString()}`), }); } @@ -480,10 +626,21 @@ async function runRollupAggregate( granularity: RollupGranularity; from: Date; to: Date; + /** + * Only the readings of these UTC days, inside `[from, to)`. A keyed + * refold (`recomputeBucketsForKeys`) passes the touched days so a handful + * spread across years does not aggregate everything in between. + */ + days?: Date[]; }, db: Pick = prisma, ): Promise { if (input.types && input.types.length === 0) return []; + // Bound as the fourth parameter, never spliced. + const dayFilter = input.days + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + : ""; + const dayParams = input.days ? [input.days.map((d) => d.toISOString())] : []; // Postgres rejects `date_trunc($1::text, m."measured_at")` in a // GROUP BY because the planner sees the parameterised expression @@ -553,6 +710,7 @@ async function runRollupAggregate( AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL + ${dayFilter} GROUP BY m."type", m."source", ${dateTrunc} `; return applyHourlyMeanDay( @@ -563,6 +721,7 @@ async function runRollupAggregate( input.userId, input.from, input.to, + ...dayParams, ), ); } @@ -600,6 +759,7 @@ async function runRollupAggregate( AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL + ${dayFilter} GROUP BY m."type", m."source", ${dateTrunc} `; return applyHourlyMeanDay( @@ -610,6 +770,7 @@ async function runRollupAggregate( input.userId, input.from, input.to, + ...dayParams, ), ); } @@ -635,6 +796,7 @@ async function applyHourlyMeanDay( granularity: RollupGranularity; from: Date; to: Date; + days?: Date[]; }, rows: RollupRow[], ): Promise { @@ -685,6 +847,11 @@ async function applyHourlyMeanDay( AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL + ${ + input.days + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + : "" + } GROUP BY m."type", m."source", date_trunc('hour', m."measured_at") ) h GROUP BY 1, 2, 3 @@ -694,6 +861,7 @@ async function applyHourlyMeanDay( input.userId, input.from, input.to, + ...(input.days ? [input.days.map((d) => d.toISOString())] : []), ); const byKey = new Map( means.map((m) => [ diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index 82e4b9e84..969bc0ee3 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -90,8 +90,13 @@ vi.mock("@/lib/integrations/status", () => ({ vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) => rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), - recomputeBucketsForMeasurement: (...a: unknown[]) => - recomputeBucketsMock(...a), + recomputeBucketsForKeys: async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) + await recomputeBucketsMock(userId, k.type, k.measuredAt); + }, })); vi.mock("@/lib/insights/comprehensive-generate", () => ({ diff --git a/src/lib/whoop/sync-core.ts b/src/lib/whoop/sync-core.ts index bb6da4b42..4ba22baeb 100644 --- a/src/lib/whoop/sync-core.ts +++ b/src/lib/whoop/sync-core.ts @@ -14,7 +14,7 @@ * `(userId, type, source = WHOOP, externalId)` so a re-post (WHOOP * re-scores recovery/sleep after the fact) overwrites in place rather than * minting a duplicate. After the upserts the rollup tier is re-folded - * (`recomputeBucketsForMeasurement`) and the status-insight caches are + * (`recomputeBucketsForKeys`) and the status-insight caches are * invalidated, identical to the Withings tail. * * The incremental window starts from `lastSyncedAt - overlap`. WHOOP re-scores @@ -37,7 +37,7 @@ import { } from "@/lib/integrations/status"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { @@ -503,9 +503,7 @@ export async function upsertWhoopMeasurements( ); try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); } catch (err) { getEvent()?.addWarning( `whoop: rollup recompute failed for ${userId}: ${err}`, diff --git a/src/lib/withings/__tests__/sync-activity.test.ts b/src/lib/withings/__tests__/sync-activity.test.ts index c9dff410a..7155da2f8 100644 --- a/src/lib/withings/__tests__/sync-activity.test.ts +++ b/src/lib/withings/__tests__/sync-activity.test.ts @@ -69,9 +69,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { const actual = await vi.importActual< typeof import("@/lib/rollups/measurement-rollups") >("@/lib/rollups/measurement-rollups"); + const perKey = vi.fn().mockResolvedValue(undefined); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForMeasurement: perKey, + // The keyed refold, observed as the per-key recomputes it stands for. + recomputeBucketsForKeys: vi.fn( + async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await perKey(userId, k.type, k.measuredAt); + }, + ), }; }); diff --git a/src/lib/withings/__tests__/sync-ecg.test.ts b/src/lib/withings/__tests__/sync-ecg.test.ts index dc8cae0b7..916677346 100644 --- a/src/lib/withings/__tests__/sync-ecg.test.ts +++ b/src/lib/withings/__tests__/sync-ecg.test.ts @@ -68,9 +68,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { const actual = await vi.importActual< typeof import("@/lib/rollups/measurement-rollups") >("@/lib/rollups/measurement-rollups"); + const perKey = vi.fn().mockResolvedValue(undefined); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForMeasurement: perKey, + // The keyed refold, observed as the per-key recomputes it stands for. + recomputeBucketsForKeys: vi.fn( + async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await perKey(userId, k.type, k.measuredAt); + }, + ), }; }); diff --git a/src/lib/withings/__tests__/sync-measure-watermark.test.ts b/src/lib/withings/__tests__/sync-measure-watermark.test.ts index 3dcb7ae9b..a242e89c0 100644 --- a/src/lib/withings/__tests__/sync-measure-watermark.test.ts +++ b/src/lib/withings/__tests__/sync-measure-watermark.test.ts @@ -16,8 +16,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { measurement: { - findFirst: vi.fn(), - create: vi.fn(), + findMany: vi.fn(), + createManyAndReturn: vi.fn(), update: vi.fn(), }, withingsConnection: { @@ -66,7 +66,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { >("@/lib/rollups/measurement-rollups"); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined), }; }); @@ -96,10 +96,6 @@ import { import { getValidToken, syncUserMeasurements } from "../sync"; -const P2002 = Object.assign(new Error("Unique constraint failed"), { - code: "P2002", -}); - beforeEach(() => { vi.clearAllMocks(); vi.mocked(prisma.$transaction).mockImplementation(async (run) => @@ -116,8 +112,15 @@ beforeEach(() => { lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000), } as never); vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation( + (async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((row, i) => ({ + id: `m-${i}`, + type: row.type, + measuredAt: row.measuredAt, + }))) as never, + ); vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); }); @@ -168,10 +171,20 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () measuredAt: new Date("2026-05-16T08:05:00Z"), }, ] as never); - // The first write hard-fails (a transient DB error), the second succeeds. - vi.mocked(prisma.measurement.create) - .mockRejectedValueOnce(new Error("connection reset by peer")) - .mockResolvedValue({} as never); + // The weight is stored with an older value; its update hard-fails (a + // transient DB error). The pulse is new and inserts. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + id: "m-weight", + type: "WEIGHT", + measuredAt: new Date("2026-05-16T08:00:00Z"), + value: 79, + deletedAt: null, + }, + ] as never); + vi.mocked(prisma.measurement.update).mockRejectedValueOnce( + new Error("connection reset by peer"), + ); const result = await syncUserMeasurements("user-1"); @@ -187,7 +200,7 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () ); }); - it("does NOT hold the watermark for a benign P2002 idempotent-write collision", async () => { + it("does NOT hold the watermark for a duplicate a concurrent run inserted first", async () => { vi.mocked(fetchMeasurements).mockResolvedValue([ { type: "WEIGHT", @@ -195,11 +208,15 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () measuredAt: new Date("2026-05-16T08:00:00Z"), }, ] as never); - vi.mocked(prisma.measurement.create).mockRejectedValueOnce(P2002); + // A concurrent run inserted the same reading first: the batched insert + // skips it as a duplicate. + vi.mocked(prisma.measurement.createManyAndReturn).mockResolvedValueOnce( + [] as never, + ); const { imported } = await syncUserMeasurements("user-1"); - // P2002 means the row is already present — nothing lost — so the cycle is + // A skipped duplicate is already present — nothing lost — so the cycle is // still healthy: stamp the watermark + success, record no failure. expect(imported).toBe(0); expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1); diff --git a/src/lib/withings/__tests__/sync-plan.test.ts b/src/lib/withings/__tests__/sync-plan.test.ts index 89834898e..1cf80a2a5 100644 --- a/src/lib/withings/__tests__/sync-plan.test.ts +++ b/src/lib/withings/__tests__/sync-plan.test.ts @@ -10,8 +10,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { measurement: { - findFirst: vi.fn(), - create: vi.fn(), + findMany: vi.fn(), + createManyAndReturn: vi.fn(), update: vi.fn(), }, withingsConnection: { @@ -61,7 +61,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { >("@/lib/rollups/measurement-rollups"); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined), }; }); @@ -179,8 +179,10 @@ function persistedLegs(): Array<{ leg: unknown; value: unknown }> { function createdTypes(): string[] { return vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0] as { data: { type: string } }).data.type); + .mocked(prisma.measurement.createManyAndReturn) + .mock.calls.flatMap((c) => + (c[0] as { data: Array<{ type: string }> }).data.map((d) => d.type), + ); } beforeEach(() => { @@ -193,14 +195,15 @@ beforeEach(() => { connection() as never, ); vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockImplementation((async (args: { - data: { type: string; measuredAt: Date }; - }) => ({ - id: `m-${args.data.type}`, - type: args.data.type, - measuredAt: args.data.measuredAt, - })) as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation( + (async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((d) => ({ + id: `m-${d.type}`, + type: d.type, + measuredAt: d.measuredAt, + }))) as never, + ); vi.mocked(prisma.$executeRaw).mockResolvedValue(1 as never); }); @@ -338,7 +341,7 @@ describe("syncUserMeasurements inside the Withings API plan", () => { for (let run = 0; run < 4; run++) { vi.mocked(prisma.$executeRaw).mockClear(); - vi.mocked(prisma.measurement.create).mockClear(); + vi.mocked(prisma.measurement.createManyAndReturn).mockClear(); requestedStartdates.length = 0; await syncUserMeasurements("user-1"); const [persisted] = persistedLegs(); diff --git a/src/lib/withings/__tests__/sync-plausibility.test.ts b/src/lib/withings/__tests__/sync-plausibility.test.ts index c980998a5..b80a6715c 100644 --- a/src/lib/withings/__tests__/sync-plausibility.test.ts +++ b/src/lib/withings/__tests__/sync-plausibility.test.ts @@ -14,10 +14,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { measurement: { - findFirst: vi.fn(), findMany: vi.fn(), - create: vi.fn(), createMany: vi.fn(), + createManyAndReturn: vi.fn(), update: vi.fn(), }, withingsConnection: { @@ -67,6 +66,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { return { ...actual, recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined), }; }); @@ -125,13 +125,15 @@ beforeEach(() => { lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000), } as never); vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); - vi.mocked(prisma.measurement.create).mockResolvedValue({ - id: "m-1", - type: "PULSE", - measuredAt: new Date(), - } as never); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation( + (async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((d, i) => ({ + id: `m-${i}`, + type: d.type, + measuredAt: d.measuredAt, + }))) as never, + ); vi.mocked(prisma.measurement.createMany).mockResolvedValue({ count: 1, } as never); @@ -157,7 +159,11 @@ describe("syncUserMeasurements — plausibility gate", () => { ); expect(result).toEqual({ imported: 1, failed: false }); - expect(prisma.measurement.create).toHaveBeenCalledTimes(1); + expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1); + expect( + vi.mocked(prisma.measurement.createManyAndReturn).mock.calls[0]?.[0] + ?.data, + ).toHaveLength(1); expect(tally()).toBeUndefined(); }); @@ -182,7 +188,11 @@ describe("syncUserMeasurements — plausibility gate", () => { // Only the real reading landed, and the run is still a clean success: one // impossible sample must not strand the rest behind a held watermark. expect(result).toEqual({ imported: 1, failed: false }); - expect(prisma.measurement.create).toHaveBeenCalledTimes(1); + expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1); + expect( + vi.mocked(prisma.measurement.createManyAndReturn).mock.calls[0]?.[0] + ?.data, + ).toHaveLength(1); expect(tally()).toEqual({ total: 1, buckets: [ @@ -196,9 +206,15 @@ describe("syncUserMeasurements — plausibility gate", () => { // The slot already holds a real pulse; the provider re-sends the same // instant carrying a decode slip. Ungated, the update branch rewrote the // good value with the bad one. - vi.mocked(prisma.measurement.findFirst).mockResolvedValue({ - id: "existing-row", - } as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValueOnce([ + { + id: "existing-row", + type: "PULSE", + measuredAt: new Date("2026-05-16T08:00:00Z"), + value: 61, + deletedAt: null, + }, + ] as never); vi.mocked(fetchMeasurements).mockResolvedValue([ { type: "PULSE", diff --git a/src/lib/withings/__tests__/sync-sleep.test.ts b/src/lib/withings/__tests__/sync-sleep.test.ts index d5bb12404..ea945c843 100644 --- a/src/lib/withings/__tests__/sync-sleep.test.ts +++ b/src/lib/withings/__tests__/sync-sleep.test.ts @@ -67,9 +67,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { const actual = await vi.importActual< typeof import("@/lib/rollups/measurement-rollups") >("@/lib/rollups/measurement-rollups"); + const perKey = vi.fn().mockResolvedValue(undefined); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForMeasurement: perKey, + // The keyed refold, observed as the per-key recomputes it stands for. + recomputeBucketsForKeys: vi.fn( + async ( + userId: string, + keys: Array<{ type: string; measuredAt: Date }>, + ) => { + for (const k of keys) await perKey(userId, k.type, k.measuredAt); + }, + ), }; }); diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index 4bcdb7d50..1a075fec6 100644 --- a/src/lib/withings/sync-activity.ts +++ b/src/lib/withings/sync-activity.ts @@ -45,7 +45,7 @@ import { getUnitForType } from "@/lib/validations/measurement"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; @@ -540,9 +540,7 @@ export async function syncUserActivity( // catch-up costs at most ~N (type, day) recomputes. try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); // v1.8.0 — drop the per-metric assessment caches the synced types // dirty (steps feed the general overview). Fire-and-forget. diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts index b4fe85fe7..865532dc5 100644 --- a/src/lib/withings/sync-ecg.ts +++ b/src/lib/withings/sync-ecg.ts @@ -56,7 +56,7 @@ import { safeFetch } from "@/lib/safe-fetch"; import { persistEcgRecording } from "@/lib/ecg/persist-recording"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; @@ -519,9 +519,7 @@ export async function syncUserEcg( // then drop the per-metric assessment caches the new events dirty. try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); invalidateStatusInsightsForTypes( userId, keys.map((k) => k.type), diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index 0611127ec..41dd7f5b1 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/src/lib/withings/sync-sleep.ts @@ -45,7 +45,7 @@ import { safeFetch } from "@/lib/safe-fetch"; import { getUnitForType } from "@/lib/validations/measurement"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; @@ -628,9 +628,7 @@ export async function syncUserSleep( // ~10 stage rows per night land in one DAY recompute. try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); // v1.8.0 — drop the per-metric assessment caches the synced types // dirty (sleep + resting-HR feed the pulse / general cards). diff --git a/src/lib/withings/sync.ts b/src/lib/withings/sync.ts index 1c84a09b0..6bf90b248 100644 --- a/src/lib/withings/sync.ts +++ b/src/lib/withings/sync.ts @@ -29,7 +29,6 @@ import { } from "./api-plan"; import { getUserWithingsCredentials } from "./credentials"; import { annotate, getEvent } from "@/lib/logging/context"; -import { isP2002 } from "@/lib/prisma-errors"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; import { @@ -45,7 +44,7 @@ import { } from "./response-classifier"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { @@ -349,7 +348,8 @@ export async function syncUserMeasurements( // through Withings. We collect first, fold once at the end so a 100- // row monthly catch-up costs at most ~N (type, day) recomputes rather // than N per-row hooks. Best-effort: a populator hiccup never fails - // the user's sync. + // the user's sync. Only readings this run inserted or changed count as + // touched: a re-read reading that is already stored dirties nothing. const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; const insertedArrivals: Array<{ id: string; @@ -367,69 +367,118 @@ export async function syncUserMeasurements( let anyRowFailed = false; let firstRowError: unknown = null; + // The run reads a window (the whole history on a full sync) and most of it + // is usually stored already. One read of the stored rows in that window + // splits it three ways: unchanged (nothing to write, nothing to refold), + // changed or tombstoned (an update), and new (one batched insert). The + // per-row `findFirst` + `update` this replaced cost two round trips per + // reading and marked every re-read day dirty, which is where a full sync's + // forty seconds went. + const latest = new Map(); for (const m of measures) { - const measType = m.type as MeasurementType; - try { - // v1.4.25 — Migration 0055 added `sleepStage` to the - // composite unique with `NULLS NOT DISTINCT`, so a non-sleep - // upsert keyed on (userId, type, measuredAt, source) with - // sleepStage IS NULL is still safe. But Prisma's typed compound - // input requires a non-null `sleepStage`, so we model the - // idempotent write as a `findFirst` + `create`/`update`. The - // unique index still serializes concurrent inserts, so the worst - // case is a Prisma P2002 we catch and ignore. - const existing = await prisma.measurement.findFirst({ - where: { - userId, - type: measType, - measuredAt: m.measuredAt, - source: "WITHINGS", - sleepStage: null, + latest.set(`${m.type}|${m.measuredAt.getTime()}`, m); + } + const incoming = [...latest.values()]; + const existingByKey = new Map< + string, + { id: string; value: number; deletedAt: Date | null } + >(); + if (incoming.length > 0) { + let first = incoming[0]!.measuredAt; + let last = first; + for (const m of incoming) { + if (m.measuredAt < first) first = m.measuredAt; + if (m.measuredAt > last) last = m.measuredAt; + } + const stored = await prisma.measurement.findMany({ + where: { + userId, + source: "WITHINGS", + sleepStage: null, + type: { + in: [...new Set(incoming.map((m) => m.type as MeasurementType))], }, - select: { id: true }, + measuredAt: { gte: first, lte: last }, + }, + select: { + id: true, + type: true, + measuredAt: true, + value: true, + deletedAt: true, + }, + }); + for (const row of stored) { + existingByKey.set(`${row.type}|${row.measuredAt.getTime()}`, row); + } + } + + const toCreate: typeof incoming = []; + for (const m of incoming) { + const measType = m.type as MeasurementType; + const existing = existingByKey.get(`${m.type}|${m.measuredAt.getTime()}`); + if (!existing) { + toCreate.push(m); + continue; + } + if (existing.value === m.value && existing.deletedAt === null) { + imported++; + continue; + } + try { + await prisma.measurement.update({ + where: { id: existing.id }, + // `deletedAt: null` is a no-op on a live row, a deliberate + // RESURRECTION on a tombstoned one — Withings is the source of + // truth for its own rows, so a re-fetched measure brings the row + // back (mirrors Google / Fitbit). + data: { value: m.value, deletedAt: null }, }); - if (existing) { - await prisma.measurement.update({ - where: { id: existing.id }, - // `deletedAt: null` is a no-op on a live row, a deliberate - // RESURRECTION on a tombstoned one — Withings is the source of - // truth for its own rows, so a re-fetched measure brings the row - // back (mirrors Google / Fitbit). - data: { value: m.value, deletedAt: null }, - }); - } else { - const created = await prisma.measurement.create({ - data: { - userId, - type: measType, - value: m.value, - unit: getUnitForType(m.type), - measuredAt: m.measuredAt, - source: "WITHINGS", - }, - select: { id: true, type: true, measuredAt: true }, - }); - insertedArrivals.push(created); - } touched.push({ type: measType, measuredAt: m.measuredAt }); imported++; } catch (err) { - if (isP2002(err)) { - // Benign: the composite unique index serialised a concurrent insert of - // this exact reading — the row is present, nothing is lost. Warn and - // continue WITHOUT holding the watermark (this is the expected - // idempotent-write race the findFirst+create models around). - getEvent()?.addWarning( - `Withings measure upsert hit an idempotent P2002 (row already present): ${err}`, - ); - } else { - // A genuine write failure strands this reading. Flag the cycle so the - // watermark is held and the next scheduled tick refetches the overlap - // window and retries. - anyRowFailed = true; - if (firstRowError === null) firstRowError = err; - getEvent()?.addWarning(`Failed to upsert measure: ${err}`); + // A genuine write failure strands this reading. Flag the cycle so the + // watermark is held and the next scheduled tick refetches the overlap + // window and retries. + anyRowFailed = true; + if (firstRowError === null) firstRowError = err; + getEvent()?.addWarning(`Failed to update measure: ${err}`); + } + } + + // New readings in one statement per chunk. `skipDuplicates` absorbs the + // idempotent-write race the unique index serialises (a concurrent webhook + // run inserting the same reading): that row is present and nothing is lost, + // exactly the P2002 the per-row path caught and ignored. + for (let i = 0; i < toCreate.length; i += 1_000) { + const chunk = toCreate.slice(i, i + 1_000); + try { + const created = await prisma.measurement.createManyAndReturn({ + data: chunk.map((m) => ({ + userId, + type: m.type as MeasurementType, + value: m.value, + unit: getUnitForType(m.type), + measuredAt: m.measuredAt, + source: "WITHINGS" as const, + })), + skipDuplicates: true, + select: { id: true, type: true, measuredAt: true }, + }); + for (const row of created) { + insertedArrivals.push(row); + touched.push({ type: row.type, measuredAt: row.measuredAt }); } + imported += created.length; + if (created.length < chunk.length) { + getEvent()?.addWarning( + `Withings measure insert skipped ${chunk.length - created.length} rows already present (idempotent race)`, + ); + } + } catch (err) { + anyRowFailed = true; + if (firstRowError === null) firstRowError = err; + getEvent()?.addWarning(`Failed to insert measures: ${err}`); } } @@ -446,9 +495,7 @@ export async function syncUserMeasurements( // chart at the 30-day range under-counted recent days. try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); // v1.8.0 — a Withings sync that lands new weight / BP / pulse / // body-comp rows dirties the matching per-metric assessment caches. diff --git a/src/proxy.ts b/src/proxy.ts index 1d54b9f39..d2f8e2e81 100644 --- a/src/proxy.ts +++ b/src/proxy.ts @@ -8,6 +8,7 @@ import { import { setBaselineSecurityHeaders } from "@/lib/http/proxy-bypass-routes"; import { DEMO_REFUSAL_BODY } from "@/lib/demo-refusal"; import { resolveRequestId } from "@/lib/http/request-id"; +import { redactSecrets } from "@/lib/logging/redact"; /** * Paths that do NOT require a session cookie (public pages + external webhooks). @@ -321,6 +322,9 @@ function applyBaselineSecurityHeaders(response: NextResponse): NextResponse { */ const NOTIFICATION_ACTION_PREFIX = "/api/notifications/action/"; +/** The request header that marks a Next.js Server Action call. */ +const SERVER_ACTION_HEADER = "next-action"; + type NotificationActionRefusal = "method_not_allowed" | "worker_only" | "demo_mode"; @@ -371,6 +375,42 @@ export function proxy(request: NextRequest) { ); } + // A Server Action call. HealthLog defines no Server Actions (no module + // carries `"use server"`, which `no-server-actions-guard.test.ts` holds), so + // no bundle of this app, old or new, ever sends `Next-Action`: the requests + // that do are scanners probing the Server Action surface, nightly and in + // bursts. Left to Next, each one printed "Failed to find Server Action" at + // error level, unstructured, an error with no user behind it. Refused here + // instead, with one countable `warn` line: a refusal by design. + if (request.headers.has(SERVER_ACTION_HEADER)) { + console.warn( + JSON.stringify({ + timestamp: new Date().toISOString(), + level: "warn", + kind: "http", + message: "Server Action call refused: this app defines none.", + action: { name: "proxy.server_action.refused" }, + http: { + method: request.method, + // A probe may aim at a path that carries a secret segment. + path: redactSecrets(pathname), + status: 404, + user_agent: request.headers.get("user-agent") ?? undefined, + }, + }), + ); + return applyBaselineSecurityHeaders( + NextResponse.json( + { + data: null, + error: "Not found", + meta: { errorCode: "server_action.unsupported" }, + }, + { status: 404 }, + ), + ); + } + // 301 redirects for renamed routes const redirect = LEGACY_REDIRECTS[pathname]; if (redirect) { diff --git a/tests/integration/codex-credential-park.test.ts b/tests/integration/codex-credential-park.test.ts new file mode 100644 index 000000000..95ea57391 --- /dev/null +++ b/tests/integration/codex-credential-park.test.ts @@ -0,0 +1,263 @@ +/** + * A Codex sign-in the issuer refused is parked until the person signs in + * again, against the real rows. + * + * Before the park, a dead refresh token was asked again on every hop of + * every run: the chain resolved Codex, the proactive refresh failed and was + * swallowed, the request 401'd, the on-401 refresh failed again, and the + * nightly warm repeated that for the briefing and every card. Parking flips + * the connection to the `expired` state Settings already renders, so the + * chain and the presence probe stop counting it and nothing leaves until a + * new sign-in writes `connected`. + */ +import { NextRequest } from "next/server"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { cookieJar, headerJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/logging/signal", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, emitSignal: vi.fn(actual.emitSignal) }; +}); + +// Only the two calls that would leave the machine are replaced; the +// credential codec stays real so the rows hold genuine ciphertext. +vi.mock("@/lib/ai/codex-oauth", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + refreshDeviceTokens: vi.fn(), + pollDeviceCode: vi.fn(), + }; +}); + +import { + CodexRefreshError, + encryptAdminCodexCreds, + encryptCodexCreds, + pollDeviceCode, + refreshDeviceTokens, +} from "@/lib/ai/codex-oauth"; +import { encrypt } from "@/lib/crypto"; +import { emitSignal } from "@/lib/logging/signal"; +import { probeProviderChain, resolveProviderChain } from "@/lib/ai/provider"; +import { postgresProviderHealthLedger } from "@/lib/ai/provider-health-ledger"; + +const prisma = getPrismaClient(); + +const HOUR_MS = 60 * 60 * 1000; + +function creds(tag: string, expiresInMs: number) { + return { + accessToken: `access-${tag}`, + refreshToken: `refresh-${tag}`, + accountId: `account-${tag}`, + expiresAt: new Date(Date.now() + expiresInMs), + }; +} + +function parkSignals(provider: string) { + return vi + .mocked(emitSignal) + .mock.calls.filter( + ([input]) => + input.action === "ai.credential.expired" && + input.meta?.provider === provider, + ); +} + +async function seedCodexUser() { + // An access token that expired long ago: the state a dead sign-in sits in. + const enc = encryptCodexCreds(creds("old", -HOUR_MS)); + const user = await prisma.user.create({ + data: { + username: "codex-park", + email: "codex-park@example.test", + aiProviderChain: [{ providerType: "codex", priority: 1, enabled: true }], + codexAccessTokenEncrypted: enc.accessEncrypted, + codexRefreshTokenEncrypted: enc.refreshEncrypted, + codexConnectionStatus: "connected", + codexConnectedAt: new Date(), + }, + }); + const session = await prisma.session.create({ + data: { userId: user.id, expiresAt: new Date(Date.now() + 60_000) }, + }); + cookieJar.set("healthlog_session", session.id); + return user; +} + +beforeEach(async () => { + await truncateAllTables(prisma); + cookieJar.clear(); + headerJar.clear(); + vi.mocked(emitSignal).mockClear(); + vi.mocked(refreshDeviceTokens).mockReset(); + vi.mocked(pollDeviceCode).mockReset(); +}); + +afterEach(() => { + vi.mocked(refreshDeviceTokens).mockReset(); +}); + +describe("a refused Codex sign-in", () => { + it("is parked once, stops being asked, and resumes after a new sign-in", async () => { + const user = await seedCodexUser(); + vi.mocked(refreshDeviceTokens).mockRejectedValue( + new CodexRefreshError(400, "invalid_grant"), + ); + + // Two workers meet the refusal at once: one transition, one signal. + const [first, second] = await Promise.all([ + resolveProviderChain(user.id), + resolveProviderChain(user.id), + ]); + expect(first.map((p) => p.providerType)).not.toContain("codex"); + expect(second.map((p) => p.providerType)).not.toContain("codex"); + expect( + ( + await prisma.user.findUniqueOrThrow({ + where: { id: user.id }, + select: { codexConnectionStatus: true }, + }) + ).codexConnectionStatus, + ).toBe("expired"); + expect(parkSignals("codex")).toHaveLength(1); + expect(parkSignals("codex")[0][0].meta?.oauth_error).toBe("invalid_grant"); + + // Parked: later runs neither ask the issuer nor count Codex as present. + const asked = vi.mocked(refreshDeviceTokens).mock.calls.length; + for (let i = 0; i < 3; i += 1) { + const chain = await resolveProviderChain(user.id); + expect(chain.map((p) => p.providerType)).not.toContain("codex"); + } + expect(vi.mocked(refreshDeviceTokens).mock.calls.length).toBe(asked); + expect(parkSignals("codex")).toHaveLength(1); + const presence = await probeProviderChain(user.id); + expect(presence.availableTypes).not.toContain("codex"); + + // The hop that met the refusal left an auth_failed row behind. + await postgresProviderHealthLedger.recordFailure(user.id, "codex", 401); + expect( + (await postgresProviderHealthLedger.getSkipHints(user.id)).get("codex") + ?.reason, + ).toBe("credential_expired"); + + // The person signs in again through the device-code flow. + vi.mocked(pollDeviceCode).mockResolvedValue({ + status: "connected", + creds: creds("new", 10 * HOUR_MS), + } as Awaited>); + cookieJar.set( + "codex_device", + encrypt(JSON.stringify({ deviceAuthId: "device", userCode: "CODE" })), + ); + const { POST } = await import("@/app/api/auth/codex/device-poll/route"); + const res = await POST( + new NextRequest("http://localhost/api/auth/codex/device-poll", { + method: "POST", + }), + ); + expect(res.status).toBe(200); + + // Resumed in place: connected, first in the chain, not benched by the + // refused credential's ledger row, and the fresh token needs no refresh. + const resumed = await resolveProviderChain(user.id); + expect(resumed[0]?.providerType).toBe("codex"); + expect( + (await postgresProviderHealthLedger.getSkipHints(user.id)).has("codex"), + ).toBe(false); + expect(vi.mocked(refreshDeviceTokens).mock.calls.length).toBe(asked); + }); + + it("is not parked by a refusal that says nothing about the credential", async () => { + const user = await seedCodexUser(); + vi.mocked(refreshDeviceTokens).mockRejectedValue( + new CodexRefreshError(503, null), + ); + + const chain = await resolveProviderChain(user.id); + + // A brown-out at the issuer: the connection stays, the client's own + // on-401 path gets its chance, and nothing is signalled. + expect(chain.map((p) => p.providerType)).toContain("codex"); + expect( + ( + await prisma.user.findUniqueOrThrow({ + where: { id: user.id }, + select: { codexConnectionStatus: true }, + }) + ).codexConnectionStatus, + ).toBe("connected"); + expect(parkSignals("codex")).toHaveLength(0); + }); + + it("parks the operator's central connection the same way", async () => { + const enc = encryptAdminCodexCreds(creds("central", -HOUR_MS)); + await prisma.appSettings.create({ + data: { + id: "singleton", + adminCodexAccessTokenEncrypted: enc.accessEncrypted, + adminCodexRefreshTokenEncrypted: enc.refreshEncrypted, + adminCodexAccountIdEncrypted: enc.accountIdEncrypted, + adminCodexTokenExpiresAt: enc.expiresAt, + adminCodexConnectionStatus: "connected", + }, + }); + const user = await prisma.user.create({ + data: { + username: "central-park", + email: "central-park@example.test", + aiProviderChain: [], + useCentralCodex: true, + }, + }); + vi.mocked(refreshDeviceTokens).mockRejectedValue( + new CodexRefreshError(401, null), + ); + + const chain = await resolveProviderChain(user.id); + expect(chain.map((p) => p.providerType)).not.toContain("admin-codex"); + expect( + ( + await prisma.appSettings.findUniqueOrThrow({ + where: { id: "singleton" }, + select: { adminCodexConnectionStatus: true }, + }) + ).adminCodexConnectionStatus, + ).toBe("expired"); + expect(parkSignals("admin-codex")).toHaveLength(1); + + await resolveProviderChain(user.id); + expect(vi.mocked(refreshDeviceTokens)).toHaveBeenCalledTimes(1); + }); +}); diff --git a/tests/integration/dashboard-summary-reads.test.ts b/tests/integration/dashboard-summary-reads.test.ts index 6dcf1f840..979c6746f 100644 --- a/tests/integration/dashboard-summary-reads.test.ts +++ b/tests/integration/dashboard-summary-reads.test.ts @@ -27,12 +27,15 @@ import { beforeEach, describe, expect, it } from "vitest"; import { getPrismaClient, truncateAllTables } from "./setup"; import { + allTimeRollupBoundary, + readAllTimeCounts, readLatestEver, readSparkBuckets, type LatestEverRow, type SparklineRow, } from "@/lib/dashboard/summary-reads"; import { measurementTypeEnum } from "@/lib/validations/measurement"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; import type { MeasurementType, RollupGranularity, @@ -245,3 +248,92 @@ describe("dashboard summary reads — parity with the pre-v1.38.5 SQL", () => { } }); }); + +describe("dashboard all-time counts — parity with the live count", () => { + /** The pre-rollup read: one index entry per live reading. */ + async function legacyAllTime(userId: string) { + const groups = await getPrismaClient().measurement.groupBy({ + by: ["type"], + where: { userId, type: { in: TYPES }, deletedAt: null }, + _count: { _all: true }, + }); + return new Map(groups.map((g) => [g.type, g._count._all])); + } + + it("counts what the live count counts, from the rollups", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { username: "all-time-counts", email: "all-time@example.test" }, + }); + const now = new Date(); + const boundary = allTimeRollupBoundary(now); + + // A watch-dense pulse history from two sources: a reading every four + // minutes for two years, written set-wise the way an import lands it. + const twoYears = 2 * 365 * DAY; + for (const [source, offsetMinutes] of [ + ["APPLE_HEALTH", 0], + ["WITHINGS", 2], + ] as const) { + await prisma.$executeRawUnsafe( + `INSERT INTO measurements + (id, user_id, type, value, unit, source, measured_at, created_at, updated_at) + SELECT md5($1 || $2 || g::text), $1, 'PULSE'::"measurement_type", + 60 + (g % 30), 'bpm', $2::"measurement_source", + $3::timestamptz + make_interval(mins => g * 4 + $4), now(), now() + FROM generate_series(0, $5::int) AS g`, + user.id, + source, + new Date(now.getTime() - twoYears), + offsetMinutes, + Math.floor(twoYears / (4 * 60_000)) - 1, + ); + } + // Weigh-ins: some older than the fold window, one tombstoned. + await prisma.measurement.createMany({ + data: [ + ...[1, 2, 3].map((years) => ({ + userId: user.id, + type: "WEIGHT" as MeasurementType, + value: 80 + years, + unit: "kg", + measuredAt: new Date(boundary.getTime() - years * 30 * DAY), + })), + ...[3, 2, 1].map((days) => ({ + userId: user.id, + type: "WEIGHT" as MeasurementType, + value: 79, + unit: "kg", + measuredAt: new Date(now.getTime() - days * DAY), + deletedAt: days === 2 ? now : null, + })), + ], + }); + await recomputeUserRollups(user.id, { granularities: ["DAY"] }); + // A type written after the fold, never bucketed: counted live. + await prisma.measurement.createMany({ + data: [5, 4].map((days) => ({ + userId: user.id, + type: "BLOOD_GLUCOSE" as MeasurementType, + value: 98, + unit: "mg/dL", + measuredAt: new Date(now.getTime() - days * DAY), + })), + }); + + let started = performance.now(); + const legacy = await legacyAllTime(user.id); + const legacyMs = performance.now() - started; + started = performance.now(); + const shipped = await readAllTimeCounts(prisma, user.id, TYPES, now); + const shippedMs = performance.now() - started; + console.log( + `all-time counts over ${legacy.get("PULSE")} pulse rows: live ${legacyMs.toFixed(0)} ms, rollups ${shippedMs.toFixed(0)} ms`, + ); + + expect(new Map(shipped.map((r) => [r.type, r.count]))).toEqual(legacy); + expect(legacy.get("PULSE")).toBeGreaterThan(500_000); + expect(legacy.get("WEIGHT")).toBe(5); + expect(legacy.get("BLOOD_GLUCOSE")).toBe(2); + }); +}); diff --git a/tests/integration/measurement-rollups.test.ts b/tests/integration/measurement-rollups.test.ts index 2d91a56e1..2e382c9a5 100644 --- a/tests/integration/measurement-rollups.test.ts +++ b/tests/integration/measurement-rollups.test.ts @@ -29,6 +29,7 @@ import { getPrismaClient, truncateAllTables } from "./setup"; import { ROLLUP_FULL_BACKFILL_QUEUE, enqueueBootTimeRollupBackfill, + recomputeBucketsForKeys, recomputeBucketsForMeasurement, recomputeUserRollups, ALL_GRANULARITIES, @@ -954,3 +955,166 @@ describe("measurement rollups — integration", () => { expect(aggregate.summaries.BLOOD_PRESSURE_SYS).toBeDefined(); }); }); + +describe("keyed DAY refold", () => { + /** The DAY tier as comparable rows, `computedAt` aside. */ + async function dayTier(userId: string) { + const rows = await getPrismaClient().measurementRollup.findMany({ + where: { userId, granularity: "DAY" }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }, { source: "asc" }], + }); + // Floating sums depend on the order the scan feeds them; the last bits + // of a double are not what the comparison is about. + return rows.map(({ computedAt: _computedAt, ...rest }) => + Object.fromEntries( + Object.entries(rest).map(([k, v]) => [ + k, + typeof v === "number" ? Number(v.toPrecision(12)) : v, + ]), + ), + ); + } + + it("leaves the tier the per-key hook leaves, touching nothing else", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { username: "keyed-refold", email: "keyed-refold@example.test" }, + }); + const at = (iso: string) => new Date(iso); + const reading = ( + type: "PULSE" | "WEIGHT", + source: "MANUAL" | "WITHINGS", + iso: string, + value: number, + ) => ({ + userId: user.id, + type, + source, + value, + unit: type === "PULSE" ? "bpm" : "kg", + measuredAt: at(iso), + }); + await prisma.measurement.createMany({ + data: [ + // Two sources on one day, pulse over several hours (its day is a + // mean of hourly means), a weight two years earlier, and a day whose + // only reading is about to be deleted. + reading("PULSE", "WITHINGS", "2026-05-10T07:10:00.000Z", 58), + reading("PULSE", "WITHINGS", "2026-05-10T07:40:00.000Z", 66), + reading("PULSE", "MANUAL", "2026-05-10T19:00:00.000Z", 71), + reading("WEIGHT", "WITHINGS", "2024-03-02T06:00:00.000Z", 82.4), + reading("WEIGHT", "WITHINGS", "2026-05-11T06:00:00.000Z", 81.0), + reading("WEIGHT", "MANUAL", "2025-01-15T06:00:00.000Z", 83.3), + ], + }); + const keys = [ + { type: "PULSE" as const, measuredAt: at("2026-05-10T07:10:00.000Z") }, + { type: "WEIGHT" as const, measuredAt: at("2024-03-02T06:00:00.000Z") }, + { type: "WEIGHT" as const, measuredAt: at("2026-05-11T06:00:00.000Z") }, + ]; + // An untouched bucket, folded once, must keep its row as it is. + await recomputeBucketsForMeasurement( + user.id, + "WEIGHT", + at("2025-01-15T06:00:00.000Z"), + ); + for (const k of keys) { + await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); + } + await prisma.measurement.updateMany({ + where: { userId: user.id, measuredAt: at("2026-05-11T06:00:00.000Z") }, + data: { deletedAt: new Date() }, + }); + await prisma.measurement.create({ + data: reading("PULSE", "WITHINGS", "2026-05-10T12:00:00.000Z", 90), + }); + const untouched = await prisma.measurementRollup.findFirstOrThrow({ + where: { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: at("2025-01-15T00:00:00.000Z"), + }, + }); + + // The per-key hook's answer, then the keyed refold's, from the same rows. + for (const k of keys) { + await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); + } + const perKey = await dayTier(user.id); + await prisma.measurementRollup.deleteMany({ + where: { userId: user.id, bucketStart: { not: untouched.bucketStart } }, + }); + + const sends: Array<{ singletonKey: string }> = []; + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn(async (_queue: string, _payload: unknown, opts: never) => { + sends.push(opts); + return "job"; + }), + } as never); + await recomputeBucketsForKeys(user.id, keys); + + expect(await dayTier(user.id)).toEqual(perKey); + expect( + perKey.some( + (r) => + (r.bucketStart as Date).getTime() === + at("2026-05-11T00:00:00Z").getTime(), + ), + ).toBe(false); + const after = await prisma.measurementRollup.findFirstOrThrow({ + where: { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: untouched.bucketStart, + }, + }); + expect(after.computedAt).toEqual(untouched.computedAt); + + // One coarse job per bucket here: no two touched buckets of one type are + // adjacent, so no run merges. + expect(sends.map((o) => o.singletonKey).sort()).toEqual( + [ + `${user.id}|PULSE|WEEK|2026-05-04T00:00:00.000Z`, + `${user.id}|PULSE|MONTH|2026-05-01T00:00:00.000Z`, + `${user.id}|PULSE|YEAR|2026-01-01T00:00:00.000Z`, + `${user.id}|WEIGHT|WEEK|2024-02-26T00:00:00.000Z`, + `${user.id}|WEIGHT|WEEK|2026-05-11T00:00:00.000Z`, + `${user.id}|WEIGHT|MONTH|2024-03-01T00:00:00.000Z`, + `${user.id}|WEIGHT|MONTH|2026-05-01T00:00:00.000Z`, + `${user.id}|WEIGHT|YEAR|2024-01-01T00:00:00.000Z`, + `${user.id}|WEIGHT|YEAR|2026-01-01T00:00:00.000Z`, + ].sort(), + ); + }); + + it("merges adjacent buckets into one ranged job", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { username: "keyed-runs", email: "keyed-runs@example.test" }, + }); + const sends: Array<{ singletonKey: string }> = []; + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn(async (_queue: string, _payload: unknown, opts: never) => { + sends.push(opts); + return "job"; + }), + } as never); + // Three consecutive weeks of a daily reading. + const keys = Array.from({ length: 21 }, (_, i) => ({ + type: "WEIGHT" as const, + measuredAt: new Date(Date.UTC(2026, 4, 4 + i, 6)), + })); + + await recomputeBucketsForKeys(user.id, keys); + + const weekKeys = sends + .map((o) => o.singletonKey) + .filter((k) => k.includes("|WEEK|")); + expect(weekKeys).toEqual([ + `${user.id}|WEIGHT|WEEK|2026-05-04T00:00:00.000Z|2026-05-25T00:00:00.000Z`, + ]); + }); +}); diff --git a/tests/integration/withings-sync-volume.test.ts b/tests/integration/withings-sync-volume.test.ts new file mode 100644 index 000000000..c2b97d9aa --- /dev/null +++ b/tests/integration/withings-sync-volume.test.ts @@ -0,0 +1,204 @@ +/** + * A Withings measure sync at the volume a full re-read of a real account + * carries: two years of daily weigh-ins (seven body-composition values each) + * and a daily blood-pressure reading, about 7 300 measures over 7 300 + * (type, day) buckets. + * + * The manual full sync re-reads the whole history, and nearly all of it is + * already stored. The run must therefore cost what CHANGED, not what was + * READ: one set-based read and write for the measures, and a rollup refold + * only for the days that gained or changed a reading. The timings are logged + * for the record; the assertions are on behaviour, so a slow CI host cannot + * fail them. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { encrypt } from "@/lib/crypto"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/withings/client", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, fetchMeasurements: vi.fn() }; +}); + +import { fetchMeasurements } from "@/lib/withings/client"; +import { syncUserMeasurements } from "@/lib/withings/sync"; + +const USER = "user-withings-volume"; +const DAY_MS = 86_400_000; +const DAYS = 730; + +const prisma = getPrismaClient(); + +const WEIGH_IN: Array<[string, number]> = [ + ["WEIGHT", 81.4], + ["BODY_FAT", 21.5], + ["FAT_MASS", 17.5], + ["FAT_FREE_MASS", 63.9], + ["MUSCLE_MASS", 60.7], + ["BONE_MASS", 3.2], + ["TOTAL_BODY_WATER", 46.1], +]; +const BP: Array<[string, number]> = [ + ["BLOOD_PRESSURE_SYS", 124], + ["BLOOD_PRESSURE_DIA", 79], + ["PULSE", 64], +]; + +/** Two years of history ending yesterday, at fixed UTC minutes. */ +function history(anchor: number) { + const out: Array<{ type: string; value: number; measuredAt: Date }> = []; + for (let d = DAYS; d >= 1; d -= 1) { + const day = anchor - d * DAY_MS; + for (const [type, base] of WEIGH_IN) { + out.push({ + type, + value: base + (d % 7) * 0.1, + measuredAt: new Date(day + 6 * 3_600_000 + 12 * 60_000), + }); + } + for (const [type, base] of BP) { + out.push({ + type, + value: base + (d % 5), + measuredAt: new Date(day + 20 * 3_600_000 + 41 * 60_000), + }); + } + } + return out; +} + +beforeEach(async () => { + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: USER, + username: "withings-volume", + email: "withings-volume@example.test", + }, + }); + await prisma.withingsConnection.create({ + data: { + userId: USER, + withingsUserId: "wu-volume", + accessToken: encrypt("access-token"), + refreshToken: encrypt("refresh-token"), + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + scope: "user.metrics,user.activity", + }, + }); +}); + +afterEach(() => { + vi.mocked(fetchMeasurements).mockReset(); +}); + +describe("Withings full sync at account volume", () => { + it("writes the history once, then re-reads it at the cost of what changed", async () => { + const anchor = Date.UTC(2026, 9, 20); + const all = history(anchor); + vi.mocked(fetchMeasurements).mockResolvedValue(all); + + let started = performance.now(); + const first = await syncUserMeasurements(USER, { fullSync: true }); + const firstMs = performance.now() - started; + expect(first).toEqual({ imported: all.length, failed: false }); + expect( + await prisma.measurement.count({ + where: { userId: USER, source: "WITHINGS" }, + }), + ).toBe(all.length); + expect( + await prisma.measurementRollup.count({ + where: { userId: USER, granularity: "DAY" }, + }), + ).toBe(DAYS * (WEIGH_IN.length + BP.length)); + + // The re-read: the same history, one corrected weigh-in, one reading + // that was deleted here and comes back, and a new reading today. + const corrected = all.find((m) => m.type === "WEIGHT")!; + const resurrected = all.find((m) => m.type === "PULSE")!; + await prisma.measurement.updateMany({ + where: { + userId: USER, + type: "PULSE", + measuredAt: resurrected.measuredAt, + }, + data: { deletedAt: new Date() }, + }); + const today = { + type: "WEIGHT", + value: 80.9, + measuredAt: new Date(anchor + 7 * 3_600_000), + }; + const reread = [ + ...all.map((m) => (m === corrected ? { ...m, value: 79.0 } : m)), + today, + ]; + vi.mocked(fetchMeasurements).mockResolvedValue(reread); + + const before = new Date(); + started = performance.now(); + const second = await syncUserMeasurements(USER, { fullSync: true }); + const secondMs = performance.now() - started; + console.log( + `withings full sync: ${all.length} measures first run ${firstMs.toFixed(0)} ms, re-read ${secondMs.toFixed(0)} ms`, + ); + + expect(second).toEqual({ imported: reread.length, failed: false }); + const stored = await prisma.measurement.findMany({ + where: { + userId: USER, + source: "WITHINGS", + OR: [ + { type: "WEIGHT", measuredAt: corrected.measuredAt }, + { type: "PULSE", measuredAt: resurrected.measuredAt }, + { type: "WEIGHT", measuredAt: today.measuredAt }, + ], + }, + select: { type: true, value: true, measuredAt: true, deletedAt: true }, + }); + expect(stored).toHaveLength(3); + expect(stored.every((m) => m.deletedAt === null)).toBe(true); + expect( + stored.find( + (m) => m.measuredAt.getTime() === corrected.measuredAt.getTime(), + )?.value, + ).toBe(79.0); + + // Only the three days that changed were refolded; the other buckets keep + // the aggregate the first run wrote. + const refolded = await prisma.measurementRollup.findMany({ + where: { + userId: USER, + granularity: "DAY", + computedAt: { gte: before }, + }, + select: { type: true, bucketStart: true, mean: true }, + }); + const dayOf = (d: Date) => + new Date(Math.floor(d.getTime() / DAY_MS) * DAY_MS); + expect( + refolded.map((r) => [r.type, r.bucketStart.toISOString()]).sort(), + ).toEqual([ + ["PULSE", dayOf(resurrected.measuredAt).toISOString()], + ["WEIGHT", dayOf(corrected.measuredAt).toISOString()], + ["WEIGHT", dayOf(today.measuredAt).toISOString()], + ]); + expect( + refolded.find( + (r) => + r.type === "WEIGHT" && + r.bucketStart.getTime() === dayOf(corrected.measuredAt).getTime(), + )?.mean, + ).toBe(79.0); + }); +});