Merge pull request #1302 from MBombeck/fix/google-health-sleep-sweep

fix(google-health): protect only a night's own segments in the sleep sweep

CHANGELOG.md

@@ -16,6 +16,13 @@
   two accounts on different plans, or the shared central connection, no
   longer overwrite each other, and a sign-in reconnected to another
   account starts fresh. (#1287)
+- **A first Google Health sync no longer stalls on the sleep history.** Each
+  night's clean-up of re-scored sleep carried every segment id of the whole
+  fetch, so a backfill of a few years sent one query per night with tens of
+  thousands of values, over the database's limit, and kept the server busy
+  building them for many minutes before each one failed with a warning. Each
+  night now protects only the segments inside it, which is all the clean-up
+  can reach. (#1195)
 - **A gateway that refuses the reasoning field is asked without it for a
   day.** The lesson held for an hour, and every expiry cost one refused
   request, a few a day on a gateway that runs the briefings unattended.

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);
+  });
+});

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(

src/lib/insights/__tests__/comparison-snapshot-window.test.ts

@@ -305,10 +305,21 @@ describe("buildComparisonSnapshotForUser — output-equivalence to the raw-avera
           ? { _avg: { value: args.where.measuredAt.lte ? 82 : 80 } }
           : { _avg: { value: null } },
     );
-    // Seven-hour nights, stored in minutes.
+    // Seven-hour nights, stored in minutes. Each at midday UTC of its own
+    // calendar day: `now - n * DAY` keeps the wall-clock hour only between
+    // two clock changes, and run late in the evening across one, two of the
+    // readings fell into the same night.
+    const today = new Date();
     measurementFindMany.mockResolvedValue(
       Array.from({ length: 5 }, (_, i) => ({
-        measuredAt: new Date(Date.now() - (i + 2) * DAY),
+        measuredAt: new Date(
+          Date.UTC(
+            today.getUTCFullYear(),
+            today.getUTCMonth(),
+            today.getUTCDate() - (i + 2),
+            12,
+          ),
+        ),
         value: 420,
         sleepStage: null,
         source: "APPLE_HEALTH",

tests/integration/dawarich-context.test.ts

@@ -304,7 +304,12 @@ describe("Dawarich daily values", () => {
       UNION ALL
       SELECT row_to_json(s)::text FROM "integration_statuses" s
     `;
-    const dump = rows.map((r) => r.row).join("\n");
+    // Clock times go first: a timestamp written at :52.52x seconds holds the
+    // latitude's digits and failed the check without any coordinate stored.
+    const dump = rows
+      .map((r) => r.row)
+      .join("\n")
+      .replace(/\d{2}:\d{2}:\d{2}\.\d+/g, "<time>");
     for (const leak of [
       "Home Sweet",
       "Office Tower",

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<typeof logContext.getEvent>);
+});
+
+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);
+});