Merge branch 'fix/v14311-integ' into release/v1.43.11

tests/integration/environment-setup.ts

@@ -1,4 +1,4 @@
-import { afterEach, inject } from "vitest";
+import { afterEach, inject, vi } from "vitest";
 
 declare module "vitest" {
   export interface ProvidedContext {
@@ -32,3 +32,24 @@ afterEach(async () => {
     await import("@/lib/logging/background-tasks");
   await settleBackgroundTasks();
 });
+
+/**
+ * The same pin the unit setup applies (see `vitest.setup.ts`): the Fitbit
+ * suites run before the Web API turndown, whatever the real date, and a test
+ * about the turndown flips `isFitbitRetired` itself. The factory imports the
+ * module lazily, so nothing of the application loads before DATABASE_URL is
+ * bridged above.
+ */
+vi.mock("@/lib/fitbit/retirement", async (importOriginal) => {
+  const actual =
+    await importOriginal<typeof import("@/lib/fitbit/retirement")>();
+  const isFitbitRetired = vi.fn((_now?: Date) => false);
+  return {
+    ...actual,
+    isFitbitRetired,
+    fitbitRetirement: vi.fn((now?: Date) => ({
+      ...actual.fitbitRetirement(now),
+      ended: isFitbitRetired(now),
+    })),
+  };
+});

tests/integration/fitbit-retirement.test.ts

@@ -0,0 +1,113 @@
+/**
+ * After Google turned the Fitbit Web API off, a Fitbit connection that has
+ * gone quiet is the provider's shutdown, not a broken pipe. Against real
+ * Postgres: the health read publishes no state for it (so no card or banner
+ * asks for a reconnect), the watchdog pages nobody, and the connection row
+ * and every imported reading stay.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/notifications/dispatcher", () => ({
+  dispatchNotification: vi.fn().mockResolvedValue({
+    dispatched: true,
+    channelsAttempted: 1,
+    channelsSucceeded: 1,
+  }),
+}));
+
+import { dispatchNotification } from "@/lib/notifications/dispatcher";
+import { runIntegrationWatchdog } from "@/lib/integrations/watchdog";
+import { loadConnectionHealth } from "@/lib/integrations/connection-health-load";
+import { recordSyncSuccess } from "@/lib/integrations/status";
+import { isFitbitRetired } from "@/lib/fitbit/retirement";
+
+const HOUR = 60 * 60 * 1000;
+const DAY = 24 * HOUR;
+const NOW = new Date("2026-11-03T12:00:00.000Z");
+const USER = "user-fitbit-retired";
+const ADMIN = "user-fitbit-retired-admin";
+const prisma = () => getPrismaClient();
+
+beforeEach(async () => {
+  await truncateAllTables(prisma());
+  await prisma().user.create({
+    data: {
+      id: ADMIN,
+      username: "fr-admin",
+      email: "admin@example.test",
+      role: "ADMIN",
+    },
+  });
+  await prisma().user.create({
+    data: { id: USER, username: "fr-subject", email: "subject@example.test" },
+  });
+  await prisma().fitbitConnection.create({
+    data: {
+      userId: USER,
+      fitbitUserId: "fb-1",
+      accessToken: "not-a-real-token",
+      refreshToken: "not-a-real-token",
+      tokenExpiresAt: new Date(NOW.getTime() - DAY),
+      lastSyncedAt: new Date(NOW.getTime() - 5 * DAY),
+    },
+  });
+  // A fortnight of daily steps that stopped with the turndown.
+  await prisma().measurement.createMany({
+    data: Array.from({ length: 14 }, (_, i) => ({
+      userId: USER,
+      type: "ACTIVITY_STEPS" as const,
+      value: 8000,
+      unit: "count",
+      source: "FITBIT" as const,
+      measuredAt: new Date(NOW.getTime() - (i + 5) * DAY),
+      externalId: `stats:day-${i}`,
+    })),
+  });
+  vi.mocked(dispatchNotification).mockClear();
+  // The watchdog has run on this instance before, so it pages on a fresh
+  // incident instead of recording what stands.
+  await prisma().integrationWatch.create({
+    data: { userId: ADMIN, integration: "withings" },
+  });
+});
+
+describe("Fitbit after the Web API turndown", () => {
+  it("still reads as a connection with a health state before the day", async () => {
+    await recordSyncSuccess(USER, "fitbit");
+    const health = await loadConnectionHealth(USER, ["fitbit"]);
+    expect(health.fitbit).not.toBeNull();
+  });
+
+  it("publishes no health, pages nobody, and keeps the row and the data", async () => {
+    await recordSyncSuccess(USER, "fitbit");
+    vi.mocked(isFitbitRetired).mockReturnValue(true);
+    try {
+      const health = await loadConnectionHealth(USER, ["fitbit"]);
+      expect(health.fitbit).toBeNull();
+
+      await runIntegrationWatchdog({ mode: "hourly", now: NOW });
+      await runIntegrationWatchdog({ mode: "daily", now: NOW });
+      expect(dispatchNotification).not.toHaveBeenCalled();
+      const watch = await prisma().integrationWatch.findUnique({
+        where: { userId_integration: { userId: USER, integration: "fitbit" } },
+      });
+      expect(watch?.alertedAt ?? null).toBeNull();
+    } finally {
+      vi.mocked(isFitbitRetired).mockReturnValue(false);
+    }
+
+    expect(
+      await prisma().fitbitConnection.count({ where: { userId: USER } }),
+    ).toBe(1);
+    expect(
+      await prisma().measurement.count({
+        where: { userId: USER, source: "FITBIT" },
+      }),
+    ).toBe(14);
+  });
+});

tests/integration/hrv-rmssd-fitbit-google-migration.test.ts

@@ -0,0 +1,270 @@
+/**
+ * Migration 0420 moves the RMSSD rows Fitbit and Google Health wrote into the
+ * SDNN series over to `HRV_RMSSD`, and clears the derived state that kept the
+ * old filing: their SDNN rollup buckets, the RMSSD buckets of the accounts
+ * whose rows moved (so the boot-time discovery folds them again), and the
+ * SDNN personal records those rows set. Apple Health SDNN and WHOOP RMSSD are
+ * left exactly as they were, and a rerun changes nothing.
+ */
+import { readFileSync } from "node:fs";
+import { resolve } from "node:path";
+import { beforeAll, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import type { Prisma } from "@/generated/prisma/client";
+import {
+  enqueueBootTimeRollupBackfill,
+  recomputeUserRollups,
+} from "@/lib/rollups/measurement-rollups";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: vi.fn(() => null),
+}));
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
+
+const prisma = getPrismaClient();
+const USER = "hrv-rmssd-migration";
+const BYSTANDER = "hrv-rmssd-migration-apple-only";
+
+const MIGRATION = readFileSync(
+  resolve(
+    process.cwd(),
+    "prisma",
+    "migrations",
+    "0420_hrv_rmssd_from_fitbit_and_google_health",
+    "migration.sql",
+  ),
+  "utf8",
+);
+
+async function runMigration(): Promise<void> {
+  // The file is one `DO` block; the comments around it are harmless.
+  await prisma.$executeRawUnsafe(MIGRATION);
+}
+
+const DAY_MS = 86_400_000;
+/** Ten days ago at 07:00 UTC, so every row sits inside the fold window. */
+const D0 =
+  Math.floor(Date.now() / DAY_MS) * DAY_MS - 10 * DAY_MS + 7 * 3_600_000;
+const dayKey = (i: number) =>
+  new Date(D0 + i * DAY_MS).toISOString().slice(0, 10);
+
+function rows(): Prisma.MeasurementCreateManyInput[] {
+  const out: Prisma.MeasurementCreateManyInput[] = [];
+  for (let i = 0; i < 4; i += 1) {
+    const at = new Date(D0 + i * DAY_MS);
+    out.push(
+      {
+        id: `fitbit-${i}`,
+        userId: USER,
+        type: "HEART_RATE_VARIABILITY",
+        source: "FITBIT",
+        unit: "ms",
+        value: 40 + i,
+        measuredAt: at,
+        externalId: `${dayKey(i)}:hrv`,
+      },
+      {
+        id: `apple-${i}`,
+        userId: USER,
+        type: "HEART_RATE_VARIABILITY",
+        source: "APPLE_HEALTH",
+        unit: "ms",
+        value: 70 + i,
+        measuredAt: new Date(at.getTime() + 3_600_000),
+      },
+      // WHOOP covers the same nights in the RMSSD series already: the moved
+      // Fitbit rows land on days that already have an RMSSD bucket.
+      {
+        id: `whoop-${i}`,
+        userId: USER,
+        type: "HRV_RMSSD",
+        source: "WHOOP",
+        unit: "ms",
+        value: 55 + i,
+        measuredAt: new Date(at.getTime() + 1_800_000),
+        externalId: `sleep-${i}:hrv_rmssd`,
+      },
+      {
+        id: `bystander-${i}`,
+        userId: BYSTANDER,
+        type: "HEART_RATE_VARIABILITY",
+        source: "APPLE_HEALTH",
+        unit: "ms",
+        value: 80 + i,
+        measuredAt: at,
+      },
+    );
+  }
+  out.push(
+    {
+      // On a night WHOOP covers too: every live moved row lands on a day that
+      // already has an RMSSD bucket, so only the dropped buckets can get the
+      // account flagged for a fold.
+      id: "google-0",
+      userId: USER,
+      type: "HEART_RATE_VARIABILITY",
+      source: "GOOGLE_HEALTH",
+      unit: "ms",
+      value: 44,
+      measuredAt: new Date(D0 + DAY_MS + 7_200_000),
+      externalId: `${dayKey(1)}:hrv`,
+    },
+    // A tombstone moves too, so the sync feed reports one type for the id.
+    {
+      id: "fitbit-deleted",
+      userId: USER,
+      type: "HEART_RATE_VARIABILITY",
+      source: "FITBIT",
+      unit: "ms",
+      value: 39,
+      measuredAt: new Date(D0 + 6 * DAY_MS),
+      externalId: `${dayKey(6)}:hrv`,
+      deletedAt: new Date(),
+    },
+  );
+  return out;
+}
+
+type RollupRow = { type: string; source: string; granularity: string };
+const rollups = (userId: string) =>
+  prisma.$queryRaw<RollupRow[]>`
+    SELECT "type"::text AS type, "source"::text AS source,
+           "granularity"::text AS granularity
+    FROM measurement_rollups
+    WHERE "user_id" = ${userId}
+    ORDER BY 1, 2, 3
+  `;
+const pairs = (all: RollupRow[]) =>
+  [...new Set(all.map((r) => `${r.type}/${r.source}`))].sort();
+
+let bystanderBefore: RollupRow[] = [];
+
+beforeAll(async () => {
+  await truncateAllTables(prisma);
+  await prisma.user.createMany({
+    data: [
+      { id: USER, username: USER, timezone: "UTC" },
+      { id: BYSTANDER, username: BYSTANDER, timezone: "UTC" },
+    ],
+  });
+  await prisma.measurement.createMany({ data: rows() });
+  await recomputeUserRollups(USER);
+  await recomputeUserRollups(BYSTANDER);
+  await prisma.personalRecord.createMany({
+    data: [
+      {
+        userId: USER,
+        metricType: "HEART_RATE_VARIABILITY",
+        direction: "MAX",
+        value: 43,
+        unit: "ms",
+        achievedAt: new Date(D0 + 3 * DAY_MS),
+        source: "FITBIT",
+        sourceMeasurementId: "fitbit-3",
+      },
+      {
+        userId: USER,
+        metricType: "HEART_RATE_VARIABILITY",
+        direction: "MAX",
+        value: 73,
+        unit: "ms",
+        achievedAt: new Date(D0 + 3 * DAY_MS + 3_600_000),
+        source: "APPLE_HEALTH",
+        sourceMeasurementId: "apple-3",
+      },
+    ],
+  });
+  bystanderBefore = await rollups(BYSTANDER);
+  await runMigration();
+}, 120_000);
+
+describe("migration 0420", () => {
+  it("moves every Fitbit and Google Health SDNN row to HRV_RMSSD, tombstones included", async () => {
+    const moved = await prisma.measurement.findMany({
+      where: { userId: USER, source: { in: ["FITBIT", "GOOGLE_HEALTH"] } },
+      select: { id: true, type: true, syncVersion: true, value: true },
+      orderBy: { id: "asc" },
+    });
+    expect(moved).toHaveLength(6);
+    for (const row of moved) {
+      expect(row.type, row.id).toBe("HRV_RMSSD");
+      // A type change must reach the phone through the sync feed.
+      expect(row.syncVersion, row.id).toBe(2);
+    }
+  });
+
+  it("leaves Apple Health SDNN and WHOOP RMSSD rows alone", async () => {
+    const apple = await prisma.measurement.findMany({
+      where: { source: "APPLE_HEALTH" },
+      select: { type: true, syncVersion: true },
+    });
+    expect(apple).toHaveLength(8);
+    expect(apple.every((r) => r.type === "HEART_RATE_VARIABILITY")).toBe(true);
+    expect(apple.every((r) => r.syncVersion === 1)).toBe(true);
+    const whoop = await prisma.measurement.findMany({
+      where: { source: "WHOOP" },
+      select: { type: true, syncVersion: true },
+    });
+    expect(whoop.every((r) => r.type === "HRV_RMSSD")).toBe(true);
+    expect(whoop.every((r) => r.syncVersion === 1)).toBe(true);
+  });
+
+  it("drops the SDNN buckets of the moved sources and the account's RMSSD buckets", async () => {
+    expect(pairs(await rollups(USER))).toEqual([
+      "HEART_RATE_VARIABILITY/APPLE_HEALTH",
+    ]);
+    expect(await rollups(BYSTANDER)).toEqual(bystanderBefore);
+  });
+
+  it("leaves the account where the boot-time discovery folds it again", async () => {
+    const send = vi.fn(async () => "job-id");
+    vi.mocked(getGlobalBoss).mockReturnValue({ send } as never);
+    try {
+      const result = await enqueueBootTimeRollupBackfill();
+      expect(result.error).toBeNull();
+      const users = send.mock.calls.map(
+        (call) => (call as unknown as [string, { userId: string }])[1].userId,
+      );
+      expect(users).toContain(USER);
+      expect(users).not.toContain(BYSTANDER);
+    } finally {
+      vi.mocked(getGlobalBoss).mockReturnValue(null as never);
+    }
+  });
+
+  it("refolds into an RMSSD series that carries both straps", async () => {
+    await recomputeUserRollups(USER);
+    expect(pairs(await rollups(USER))).toEqual([
+      "HEART_RATE_VARIABILITY/APPLE_HEALTH",
+      "HRV_RMSSD/FITBIT",
+      "HRV_RMSSD/GOOGLE_HEALTH",
+      "HRV_RMSSD/WHOOP",
+    ]);
+  });
+
+  it("deletes the SDNN records the moved rows set and keeps the rest", async () => {
+    const records = await prisma.personalRecord.findMany({
+      where: { userId: USER },
+      select: { source: true, metricType: true },
+    });
+    expect(records).toEqual([
+      { source: "APPLE_HEALTH", metricType: "HEART_RATE_VARIABILITY" },
+    ]);
+  });
+
+  it("is idempotent", async () => {
+    const measurementsBefore = await prisma.measurement.findMany({
+      orderBy: { id: "asc" },
+    });
+    const rollupsBefore = await rollups(USER);
+    await runMigration();
+    expect(
+      await prisma.measurement.findMany({ orderBy: { id: "asc" } }),
+    ).toEqual(measurementsBefore);
+    expect(await rollups(USER)).toEqual(rollupsBefore);
+  });
+});

tests/integration/source-ladder-new-sources.test.ts

@@ -0,0 +1,128 @@
+/**
+ * The sources that had no place on a ladder, against real Postgres through
+ * the SQL collapse every 90-day aggregate reads (`canonicalMeasurementsFrom`).
+ *
+ * A source a ladder does not name ranks 90, below everything it does name:
+ * a weight typed as a reply to a Telegram reminder lost its day to a WHOOP
+ * strap's estimate, and a Health Connect day lost to a hand-typed one. The
+ * hand-entered sources now rank with MANUAL, Health Connect with Apple
+ * Health, and a day between two sources that were on the ladder before
+ * resolves exactly as it did.
+ */
+import { beforeAll, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import type {
+  MeasurementSource,
+  MeasurementType,
+} from "@/generated/prisma/client";
+import {
+  buildSourceRankCase,
+  canonicalMeasurementsFrom,
+} from "@/lib/analytics/source-rank-sql";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const prisma = getPrismaClient();
+const USER = "ladder-new-sources";
+const DAY = 86_400_000;
+const today = Math.floor(Date.now() / DAY) * DAY;
+
+/** One reading per (type, source) on day `n` days back, at 08:00 / 09:00 UTC. */
+function reading(
+  type: MeasurementType,
+  source: MeasurementSource,
+  n: number,
+  value: number,
+  hour = 8,
+) {
+  return {
+    userId: USER,
+    type,
+    source,
+    value,
+    unit: type === "WEIGHT" ? "kg" : "count",
+    measuredAt: new Date(today - n * DAY + hour * 3_600_000),
+  };
+}
+
+async function canonicalSources(): Promise<Map<string, string[]>> {
+  const rank = buildSourceRankCase(null, '"type"', '"source"');
+  const rows = await prisma.$queryRawUnsafe<
+    Array<{ type: string; day: Date; source: string }>
+  >(
+    `SELECT DISTINCT m."type"::text AS type,
+            date_trunc('day', m."measured_at") AS day,
+            m."source"::text AS source
+     FROM ${canonicalMeasurementsFrom(rank, "90 days")}
+     ORDER BY 1, 2, 3`,
+    USER,
+  );
+  const out = new Map<string, string[]>();
+  for (const r of rows) {
+    const key = `${r.type}@${r.day.toISOString().slice(0, 10)}`;
+    out.set(key, [...(out.get(key) ?? []), r.source]);
+  }
+  return out;
+}
+
+const dayKey = (n: number) =>
+  new Date(today - n * DAY).toISOString().slice(0, 10);
+
+beforeAll(async () => {
+  await truncateAllTables(prisma);
+  await prisma.user.create({
+    data: { id: USER, username: USER, timezone: "UTC" },
+  });
+  await prisma.measurement.createMany({
+    data: [
+      // Day 1: a strap's weight estimate and a typed reply to a reminder.
+      reading("WEIGHT", "WHOOP", 1, 84),
+      reading("WEIGHT", "TELEGRAM", 1, 82.5, 9),
+      // Day 2: an assistant logging on the person's word beside a Fitbit
+      // scale-less estimate.
+      reading("WEIGHT", "FITBIT", 2, 85),
+      reading("WEIGHT", "MCP", 2, 82.4, 9),
+      // Day 3: a hand-typed weight and the person's own bridged scale.
+      reading("WEIGHT", "MANUAL", 3, 82.3),
+      reading("WEIGHT", "EXTERNAL", 3, 82.6, 9),
+      // Day 4: two sources the ladder always named: unchanged.
+      reading("WEIGHT", "APPLE_HEALTH", 4, 82.2),
+      reading("WEIGHT", "WITHINGS", 4, 82.1, 9),
+      // Day 1: Health Connect's daily steps beside a hand-typed count.
+      reading("ACTIVITY_STEPS", "HEALTH_CONNECT", 1, 9000),
+      reading("ACTIVITY_STEPS", "MANUAL", 1, 5000, 9),
+      // Day 2: Apple Health still outranks Health Connect.
+      reading("ACTIVITY_STEPS", "APPLE_HEALTH", 2, 9100),
+      reading("ACTIVITY_STEPS", "HEALTH_CONNECT", 2, 8900, 9),
+    ],
+  });
+}, 120_000);
+
+describe("source ladders for the sources that had no place", () => {
+  it("ranks a reading the person entered through Telegram or MCP above a device estimate", async () => {
+    const picked = await canonicalSources();
+    expect(picked.get(`WEIGHT@${dayKey(1)}`)).toEqual(["TELEGRAM"]);
+    expect(picked.get(`WEIGHT@${dayKey(2)}`)).toEqual(["MCP"]);
+  });
+
+  it("keeps MANUAL above the person's own bridge", async () => {
+    const picked = await canonicalSources();
+    expect(picked.get(`WEIGHT@${dayKey(3)}`)).toEqual(["MANUAL"]);
+  });
+
+  it("ranks Health Connect right after Apple Health", async () => {
+    const picked = await canonicalSources();
+    expect(picked.get(`ACTIVITY_STEPS@${dayKey(1)}`)).toEqual([
+      "HEALTH_CONNECT",
+    ]);
+    expect(picked.get(`ACTIVITY_STEPS@${dayKey(2)}`)).toEqual(["APPLE_HEALTH"]);
+  });
+
+  it("resolves a day between two long-ranked sources exactly as before", async () => {
+    const picked = await canonicalSources();
+    expect(picked.get(`WEIGHT@${dayKey(4)}`)).toEqual(["WITHINGS"]);
+  });
+});

tests/integration/withings-sleep-sync.test.ts

@@ -51,26 +51,70 @@ afterEach(() => {
   vi.restoreAllMocks();
 });
 
-function stubWithingsSleep(series: unknown[]) {
+type Segment = {
+  startdate: number;
+  enddate: number;
+  state: number;
+  id: number;
+};
+
+/**
+ * Answer Sleep v2 the way Withings does: `getsummary` lists the nights,
+ * `get` returns the segments inside the span asked for, at most its first
+ * 24 hours.
+ */
+function stubWithingsSleep(series: Segment[]) {
+  const nights = new Map<
+    number,
+    { id: number; startdate: number; enddate: number }
+  >();
+  for (const seg of series) {
+    const night = nights.get(seg.id);
+    nights.set(seg.id, {
+      id: seg.id,
+      startdate: Math.min(night?.startdate ?? seg.startdate, seg.startdate),
+      enddate: Math.max(night?.enddate ?? seg.enddate, seg.enddate),
+    });
+  }
   vi.stubGlobal(
     "fetch",
-    vi.fn(async () => ({
-      status: 200,
-      json: async () => ({
-        status: 0,
-        body: { series },
-      }),
-    })),
+    vi.fn(async (_url: string, init: { body: string }) => {
+      const params = new URLSearchParams(init.body);
+      if (params.get("action") === "getsummary") {
+        return {
+          status: 200,
+          json: async () => ({
+            status: 0,
+            body: { series: [...nights.values()] },
+          }),
+        };
+      }
+      const from = Number(params.get("startdate"));
+      const to = Math.min(Number(params.get("enddate")), from + 24 * 3600);
+      return {
+        status: 200,
+        json: async () => ({
+          status: 0,
+          body: {
+            series: series.filter(
+              (seg) => seg.startdate >= from && seg.enddate <= to,
+            ),
+          },
+        }),
+      };
+    }),
   );
 }
 
+/** Last night: it started ten hours ago, so a run reads it again for a re-score. */
+const base = Math.floor(Date.now() / 1000) - 10 * 3600;
+
 describe("syncUserSleep — integration", () => {
   it("persists one Measurement row per stage segment for a single night", async () => {
     // Sleep session #99: 4 segments — CORE 60m, DEEP 30m, REM 30m, AWAKE 10m.
     // All segments share the same parent night but each carries a
     // distinct sleepStage, so the W17b/c composite (Migration 0055)
     // keeps every row.
-    const base = 1715000000; // 2024-05-06 13:33:20 UTC (arbitrary)
     stubWithingsSleep([
       { startdate: base, enddate: base + 3600, state: 1, id: 99 },
       { startdate: base + 3600, enddate: base + 5400, state: 2, id: 99 },
@@ -106,7 +150,6 @@ describe("syncUserSleep — integration", () => {
   });
 
   it("re-syncing the same night updates rows in place (no duplicates)", async () => {
-    const base = 1715000000;
     // First sync: DEEP 30 minutes.
     stubWithingsSleep([
       { startdate: base, enddate: base + 1800, state: 2, id: 100 },