diff --git a/src/lib/google-health/__tests__/sync-sleep-keep-ids.test.ts b/src/lib/google-health/__tests__/sync-sleep-keep-ids.test.ts new file mode 100644 index 000000000..75435af29 --- /dev/null +++ b/src/lib/google-health/__tests__/sync-sleep-keep-ids.test.ts @@ -0,0 +1,113 @@ +/** + * Pins what each sleep window protects (`protectFreshRowsInWindows`): every + * fresh row of the fetch that lies inside the window, from any session, and + * nothing outside it. Handing every window the whole fetch's ids made a full + * backfill send one query per night carrying every segment id of the history. + */ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ prisma: {} })); +vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() })); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: vi.fn(async () => false), + recordSyncFailure: vi.fn(async () => {}), + recordSyncSuccess: vi.fn(async () => {}), +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: vi.fn(() => []), + recomputeBucketsForMeasurement: vi.fn(async () => {}), + recomputeUserRollups: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/status-invalidation", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); +vi.mock("../credentials", () => ({ + getUserGoogleHealthCredentials: vi.fn(async () => null), +})); +vi.mock("@/lib/tz/resolver", () => ({ + resolveUserTimezone: vi.fn(async () => "Europe/Berlin"), +})); +vi.mock("@/lib/daily/morning-refresh-trigger", () => ({ + maybeEnqueueMorningRefresh: vi.fn(async () => {}), +})); + +import { protectFreshRowsInWindows } from "../sync-sleep"; + +const at = (iso: string) => new Date(iso); + +describe("protectFreshRowsInWindows", () => { + it("protects another session's fresh row inside the window", () => { + const [main] = protectFreshRowsInWindows( + [ + { + windowStart: at("2026-10-01T22:00:00Z"), + windowEnd: at("2026-10-02T06:00:00Z"), + keepIds: ["main:a"], + }, + ], + [ + { externalId: "main:a", measuredAt: at("2026-10-01T23:00:00Z") }, + // A nap session reported over the same stretch of the night. + { externalId: "nap:b", measuredAt: at("2026-10-02T03:00:00Z") }, + ], + ); + expect(new Set(main!.keepIds)).toEqual(new Set(["main:a", "nap:b"])); + }); + + it("leaves out fresh rows from other nights, edges included", () => { + const [w] = protectFreshRowsInWindows( + [ + { + windowStart: at("2026-10-01T22:00:00Z"), + windowEnd: at("2026-10-02T06:00:00Z"), + keepIds: ["own:a"], + }, + ], + [ + { externalId: "before", measuredAt: at("2026-10-01T21:59:59Z") }, + { externalId: "start", measuredAt: at("2026-10-01T22:00:00Z") }, + { externalId: "own:a", measuredAt: at("2026-10-02T01:00:00Z") }, + { externalId: "end", measuredAt: at("2026-10-02T06:00:00Z") }, + { externalId: "after", measuredAt: at("2026-10-02T06:00:01Z") }, + ], + ); + expect(new Set(w!.keepIds)).toEqual(new Set(["own:a", "start", "end"])); + }); + + it("passes a window without bounds through unchanged", () => { + const open = { windowStart: null, windowEnd: null, keepIds: [] }; + expect( + protectFreshRowsInWindows( + [open], + [{ externalId: "x", measuredAt: at("2026-10-01T23:00:00Z") }], + ), + ).toEqual([open]); + }); + + it("keeps each night's set to that night on a five-year backfill", () => { + const nights = 5 * 365; + const windows = []; + const readings = []; + for (let n = 0; n < nights; n++) { + const start = Date.UTC(2021, 0, 1, 22) + n * 86_400_000; + const ids: string[] = []; + for (let s = 0; s < 30; s++) { + const id = `night${n}:sleep:${s}`; + ids.push(id); + readings.push({ + externalId: id, + measuredAt: new Date(start + (s + 1) * 16 * 60_000), + }); + } + windows.push({ + windowStart: new Date(start), + windowEnd: new Date(start + 8 * 3_600_000), + keepIds: ids, + }); + } + const out = protectFreshRowsInWindows(windows, readings); + expect(out).toHaveLength(nights); + // Before, every one of the 1825 queries carried all 54750 ids. + expect(Math.max(...out.map((w) => w.keepIds.length))).toBe(30); + }); +}); diff --git a/src/lib/google-health/sync-sleep.ts b/src/lib/google-health/sync-sleep.ts index 32ff8dd8e..0e0735924 100644 --- a/src/lib/google-health/sync-sleep.ts +++ b/src/lib/google-health/sync-sleep.ts @@ -50,6 +50,55 @@ export async function syncUserSleep( ); } +/** + * Each window's keep-set: every fresh row of the fetch that lies inside the + * window, whichever session produced it. Google can report two sessions over + * the same stretch of a night (two sources, or a nap inside the main sleep), + * and a window that protected only its own session's rows tombstoned the + * other session's rows on every sync, for the upsert to bring them straight + * back. + * + * Only rows inside the window, not the whole fetch: the sweep can only reach + * rows whose `measuredAt` lies in the window, so a fresh id from another + * night protects nothing there. Handing every window the whole fetch's ids + * made a full backfill send one query per night, each carrying every segment + * id of the history, tens of thousands of bound values per query, over + * Postgres's limit and blocking the process while each one was built. A + * stored row whose fresh copy moved out of the window (a re-score shifted the + * segment) is tombstoned here and revived by the upsert that follows, which + * matches tombstoned rows by externalId. + */ +export function protectFreshRowsInWindows( + windows: GoogleHealthSleepReplaceWindow[], + readings: Array<{ externalId: string; measuredAt: Date }>, +): GoogleHealthSleepReplaceWindow[] { + const byTime = readings + .map((r) => ({ id: r.externalId, at: r.measuredAt.getTime() })) + .sort((a, b) => a.at - b.at); + // First index whose time is at or after `t`. + const lowerBound = (t: number) => { + let lo = 0; + let hi = byTime.length; + while (lo < hi) { + const mid = (lo + hi) >> 1; + if (byTime[mid]!.at < t) lo = mid + 1; + else hi = mid; + } + return lo; + }; + return windows.map((w) => { + if (!w.windowStart || !w.windowEnd) return w; + const startMs = w.windowStart.getTime(); + const endMs = w.windowEnd.getTime(); + const keep = new Set(w.keepIds); + for (let i = lowerBound(startMs); i < byTime.length; i++) { + if (byTime[i]!.at > endMs) break; + keep.add(byTime[i]!.id); + } + return { ...w, keepIds: [...keep] }; + }); +} + async function syncSleepSessions( userId: string, opts: GoogleHealthResourceSyncOptions, @@ -126,15 +175,9 @@ async function syncSleepSessions( }); } - // Every session's fresh rows are protected in every window, not only in - // their own. Google can report two sessions over the same stretch of a - // night (two sources, or a nap inside the main sleep), and a window that - // protected only its own session's rows tombstoned the other session's rows - // on every sync, for the upsert to bring them straight back. - const allKeepIds = readings.map((r) => r.externalId); await replaceStaleGoogleHealthSleep( userId, - replaceWindows.map((w) => ({ ...w, keepIds: allKeepIds })), + protectFreshRowsInWindows(replaceWindows, readings), ); const { imported, inserted } = await upsertGoogleHealthMeasurements( diff --git a/tests/integration/google-health-sleep-sweep.test.ts b/tests/integration/google-health-sleep-sweep.test.ts new file mode 100644 index 000000000..8f78bc57c --- /dev/null +++ b/tests/integration/google-health-sleep-sweep.test.ts @@ -0,0 +1,119 @@ +/** + * The Google Health sleep sweep on a full backfill, against Postgres. + * + * A first sync reads the whole sleep history, one session per night. Each + * night's window used to carry every fresh segment id of the fetch, so five + * years sent 1825 queries with 54750 bound values each: over Postgres's limit + * of 32767, every one failed with a warning and the process was busy building + * them for minutes. With each window protecting only the fresh rows inside it + * (`protectFreshRowsInWindows`), the same fetch sweeps without a warning, a + * stale row a re-score left behind is tombstoned, and a row another session + * reported over the same night survives. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { protectFreshRowsInWindows } from "@/lib/google-health/sync-sleep"; +import { replaceStaleGoogleHealthSleep } from "@/lib/google-health/sync-core"; +import * as logContext from "@/lib/logging/context"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER_ID = "user-google-sleep-sweep"; +const NIGHTS = 5 * 365; +const SEGMENTS = 30; +const DAY = 86_400_000; +const FIRST_NIGHT = Date.UTC(2021, 0, 1, 22); + +const segmentId = (night: number, s: number) => `night${night}:sleep:${s}`; +const segmentEnd = (night: number, s: number) => + new Date(FIRST_NIGHT + night * DAY + (s + 1) * 16 * 60_000); + +function backfill() { + const windows = []; + const readings = []; + for (let n = 0; n < NIGHTS; n++) { + const ids: string[] = []; + for (let s = 0; s < SEGMENTS; s++) { + ids.push(segmentId(n, s)); + readings.push({ + externalId: segmentId(n, s), + measuredAt: segmentEnd(n, s), + }); + } + windows.push({ + windowStart: new Date(FIRST_NIGHT + n * DAY), + windowEnd: new Date(FIRST_NIGHT + n * DAY + 8 * 3_600_000), + keepIds: ids, + }); + } + return { windows, readings }; +} + +async function seedSleepRow(externalId: string, measuredAt: Date) { + await getPrismaClient().measurement.create({ + data: { + userId: USER_ID, + type: "SLEEP_DURATION", + source: "GOOGLE_HEALTH", + value: 16, + unit: "min", + measuredAt, + externalId, + sleepStage: "CORE", + }, + }); +} + +let warnings: string[]; + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { id: USER_ID, username: "google-sleep-sweep" }, + }); + warnings = []; + vi.spyOn(logContext, "getEvent").mockReturnValue({ + addWarning: (w: string) => warnings.push(w), + } as unknown as ReturnType); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("the Google Health sleep sweep on a five-year backfill", () => { + it("sweeps every night without a warning, and keeps what another session reported", async () => { + const night = 1000; + // A segment from an earlier scoring that the fresh fetch no longer has. + await seedSleepRow("old-scoring:sleep:x", segmentEnd(night, 3)); + // A row another session reported over the same night, fresh in this fetch. + await seedSleepRow("nap:sleep:y", segmentEnd(night, 5)); + // A fresh segment already stored from an earlier sync. + await seedSleepRow(segmentId(night, 7), segmentEnd(night, 7)); + + const { windows, readings } = backfill(); + readings.push({ + externalId: "nap:sleep:y", + measuredAt: segmentEnd(night, 5), + }); + + const removed = await replaceStaleGoogleHealthSleep( + USER_ID, + protectFreshRowsInWindows(windows, readings), + ); + + expect(warnings).toEqual([]); + expect(removed).toBe(1); + const rows = await getPrismaClient().measurement.findMany({ + where: { userId: USER_ID }, + select: { externalId: true, deletedAt: true }, + orderBy: { externalId: "asc" }, + }); + expect(rows.map((r) => [r.externalId, r.deletedAt !== null])).toEqual([ + ["nap:sleep:y", false], + [segmentId(night, 7), false], + ["old-scoring:sleep:x", true], + ]); + }, 120_000); +});