From 07d70cd27556a0547cb91ee9bbc4461328380080 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 20:04:21 +0200 Subject: [PATCH 1/6] fix(ai): park a Codex sign-in the issuer refused until the next sign-in A dead refresh token was asked again on every hop of every run: 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. A refusal that proves the grant dead now flips the connection to the existing expired state (compare-and-set on the refused token, one signal per transition), so the chain, the presence probe and the AI capabilities stop counting it. A new device-code sign-in clears the refused credential's ledger rows so the fresh one resumes in place. --- .../admin/central-codex/device-poll/route.ts | 4 + src/app/api/auth/codex/device-poll/route.ts | 4 + src/lib/ai/provider-health-ledger.ts | 27 ++ src/lib/ai/provider.ts | 147 +++++++++- src/lib/jobs/insight-pregenerate.ts | 3 +- .../integration/codex-credential-park.test.ts | 263 ++++++++++++++++++ 6 files changed, 437 insertions(+), 11 deletions(-) create mode 100644 tests/integration/codex-credential-park.test.ts 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/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/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/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); + }); +}); From 8cdb53cb0e144411084c43d0f9abda755c75b9f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 20:13:13 +0200 Subject: [PATCH 2/6] perf(withings): write a measure sync in sets and refold only what changed A manual full sync re-reads the whole history, and nearly all of it is already stored. Each reading cost a findFirst plus an update, and every re-read day was marked dirty, so the run refolded the DAY tier and queued three coarse recomputes for every day of the account: about thirty seconds of database work behind 1.4 s of Withings calls. The measures leg now reads the stored rows of the window once, skips the unchanged ones, updates the changed or tombstoned ones and inserts the new ones in batches. Only inserted or changed readings count as touched. recomputeBucketsForKeys refolds exactly the touched DAY buckets in one aggregate and one transaction, and queues one coarse job per run of adjacent buckets instead of three per day. --- src/lib/rollups/measurement-rollups.ts | 170 ++++++++++++++- .../__tests__/sync-measure-watermark.test.ts | 49 +++-- src/lib/withings/__tests__/sync-plan.test.ts | 31 +-- .../__tests__/sync-plausibility.test.ts | 42 ++-- src/lib/withings/sync.ts | 171 +++++++++------ tests/integration/measurement-rollups.test.ts | 164 ++++++++++++++ .../integration/withings-sync-volume.test.ts | 204 ++++++++++++++++++ 7 files changed, 725 insertions(+), 106 deletions(-) create mode 100644 tests/integration/withings-sync-volume.test.ts 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/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/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/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); + }); +}); From f01f53a1cc491a5d8662e3591b994a5c9036659c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 20:20:00 +0200 Subject: [PATCH 3/6] perf(rollups): refold the buckets of every provider sync in one pass Every provider sync and the shared mutation tail refolded its touched (type, day) buckets one at a time: an aggregate, a transaction and three queue sends per day. They now hand the whole set to recomputeBucketsForKeys, which leaves the DAY tier exactly as the per-key hook did (pinned by a parity test against it) at the cost of one aggregate per chunk of keys and one coarse job per run of buckets. --- src/__tests__/measurement-write-tail-guard.test.ts | 4 +++- src/app/api/import/__tests__/route.test.ts | 12 +++++++++++- src/app/api/import/csv/__tests__/route.test.ts | 12 +++++++++++- .../measurements/bulk-delete/__tests__/route.test.ts | 12 +++++++++++- .../api/measurements/restore/__tests__/route.test.ts | 12 +++++++++++- src/lib/fitbit/sync-core.ts | 6 ++---- src/lib/google-health/sync-core.ts | 6 ++---- src/lib/nightscout/__tests__/sync.test.ts | 7 ++++++- src/lib/nightscout/sync.ts | 6 ++---- src/lib/oura/__tests__/sync.test.ts | 7 ++++++- src/lib/oura/sync.ts | 6 ++---- src/lib/polar/sync.ts | 6 ++---- .../__tests__/after-measurement-mutation.test.ts | 9 +++++++-- src/lib/rollups/after-measurement-mutation.ts | 8 +++----- src/lib/whoop/__tests__/sync.test.ts | 9 +++++++-- src/lib/whoop/sync-core.ts | 8 +++----- src/lib/withings/__tests__/sync-activity.test.ts | 12 +++++++++++- src/lib/withings/__tests__/sync-ecg.test.ts | 12 +++++++++++- src/lib/withings/__tests__/sync-sleep.test.ts | 12 +++++++++++- src/lib/withings/sync-activity.ts | 6 ++---- src/lib/withings/sync-ecg.ts | 6 ++---- src/lib/withings/sync-sleep.ts | 6 ++---- 22 files changed, 128 insertions(+), 56 deletions(-) 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/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/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/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/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/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-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). From 05ae8b1e995fc408f54653ade00276b944fcf2bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 20:27:32 +0200 Subject: [PATCH 4/6] perf(dashboard): count all-time readings from the DAY rollups The summary's all-time aggregate grouped every live reading of the account by type: one index entry per reading, millions on an account with a watch, two to four seconds whenever they were not in memory and the slowest read of the request on every cache miss. The count now sums the DAY buckets that already hold a per-day count of every source, counts live only the readings before the fold window and any type not folded yet, and takes lastSeenAt from the newest reading the latest-reading read already returns. Parity with the live count is pinned against a real Postgres. --- .../dashboard-summary-read-plan-guard.test.ts | 2 +- .../dense-measurement-read-guard.test.ts | 2 + .../dashboard/summary/__tests__/route.test.ts | 85 ++++++++--------- src/app/api/dashboard/summary/route.ts | 25 ++--- src/lib/dashboard/summary-reads.ts | 77 ++++++++++++++++ .../dashboard-summary-reads.test.ts | 92 +++++++++++++++++++ 6 files changed, 229 insertions(+), 54 deletions(-) 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/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/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/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); + }); +}); From 0534123d0d947453e8bac6a90d8ad37f1be00691 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 20:28:21 +0200 Subject: [PATCH 5/6] perf(integrations): read the status envelope in one round The provider rows, the unified connection health and the per-metric freshness were awaited one after another although none depends on the others, so a cold first visit to the settings card paid their latencies end to end. They now run in one round. --- .../status/__tests__/route.test.ts | 23 ++ src/app/api/integrations/status/route.ts | 251 +++++++++--------- 2 files changed, 153 insertions(+), 121 deletions(-) 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; From 7759729aca71d124030e2a121a874476c1c6a638 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 20:29:46 +0200 Subject: [PATCH 6/6] fix(proxy): refuse Server Action calls the app never makes HealthLog defines no Server Actions, so no bundle of it, current or stale, sends a Next-Action header. The requests that do arrive come from the nightly scan probing that surface, and Next answered each with an unstructured "Failed to find Server Action" at error level. The proxy now refuses them with a 404 and one structured warn line, and a guard keeps "use server" out of the tree so the refusal stays safe. --- src/__tests__/no-server-actions-guard.test.ts | 37 +++++++++ src/__tests__/proxy-server-action.test.ts | 77 +++++++++++++++++++ src/proxy.ts | 40 ++++++++++ 3 files changed, 154 insertions(+) create mode 100644 src/__tests__/no-server-actions-guard.test.ts create mode 100644 src/__tests__/proxy-server-action.test.ts 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/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) {