feat(withings): sleep v2 sync — stage-level segments per night

New `syncUserSleep` routine pulls per-segment sleep data from
POST /v2/sleep?action=get and writes one Measurement row per stage
segment, tagged with the appropriate SleepStage enum:

  state 0 (awake)       → AWAKE
  state 1 (light sleep) → CORE  (HealthKit-aligned NREM 1+2)
  state 2 (deep sleep)  → DEEP
  state 3 (REM)         → REM
  state 4 (synthetic)   → ignored

The legacy ASLEEP and IN_BED enum slots stay reserved for HealthKit
ingest (pre-iOS-16 `asleepUnspecified` and iOS-16+ `inBed`).
ScanWatch nights with no REM segment simply omit the row rather than
synthesising an all-zeros entry — Withings drops the field, the writer
never sees it.

Idempotency rides on Migration 0055's five-column composite — a re-sync
of the same night UPDATEs every segment row in place rather than
inserting duplicates, even when Withings re-aggregates the night with
adjusted segment boundaries.

Sleep durations stored as canonical minutes (seconds ÷ 60, rounded) so
re-syncs don't flip values on floating-point noise.

src/lib/withings/__tests__/sync-sleep.test.ts

@@ -0,0 +1,256 @@
+/**
+ * v1.4.25 W17c — Withings Sleep v2 sync unit tests.
+ *
+ * Coverage focuses on the state → SleepStage enum mapping, the
+ * unix-seconds vs minutes conversion (easy regression trap), and the
+ * per-segment write path. End-to-end DB exercise lives in the
+ * integration suite.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    measurement: {
+      findFirst: vi.fn(),
+      create: vi.fn(),
+      update: vi.fn(),
+    },
+    withingsConnection: { findUnique: vi.fn() },
+  },
+}));
+
+vi.mock("@/lib/integrations/status", () => ({
+  isReauthRequired: vi.fn().mockResolvedValue(false),
+  recordSyncFailure: vi.fn(),
+  recordSyncSuccess: vi.fn(),
+}));
+
+vi.mock("../sync", async () => {
+  const actual = await vi.importActual<typeof import("../sync")>("../sync");
+  return {
+    ...actual,
+    getValidToken: vi.fn(async () => ({
+      accessToken: "token",
+      connection: { id: "conn-1", withingsUserId: "wu-1" },
+    })),
+  };
+});
+
+vi.mock("@/lib/logging/context", () => ({
+  getEvent: vi.fn(() => ({
+    addExternalCall: vi.fn(),
+    addWarning: vi.fn(),
+  })),
+}));
+
+import { prisma } from "@/lib/db";
+import { recordSyncSuccess } from "@/lib/integrations/status";
+
+import {
+  fetchWithingsSleep,
+  mapWithingsSleepState,
+  syncUserSleep,
+} from "../sync-sleep";
+
+interface FakeSegment {
+  startdate: number;
+  enddate: number;
+  state: number;
+  id?: number;
+}
+
+function installFetchMock(segments: FakeSegment[]) {
+  const fetchMock = vi.fn(async () => ({
+    status: 200,
+    json: async () => ({
+      status: 0,
+      body: { series: segments },
+    }),
+  }));
+  vi.stubGlobal("fetch", fetchMock);
+  return fetchMock;
+}
+
+beforeEach(() => {
+  vi.resetAllMocks();
+});
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+});
+
+describe("mapWithingsSleepState", () => {
+  it("maps state 0 (awake) → AWAKE", () => {
+    expect(mapWithingsSleepState(0)).toBe("AWAKE");
+  });
+
+  it("maps state 1 (light) → CORE (HealthKit-aligned NREM 1+2)", () => {
+    expect(mapWithingsSleepState(1)).toBe("CORE");
+  });
+
+  it("maps state 2 (deep) → DEEP", () => {
+    expect(mapWithingsSleepState(2)).toBe("DEEP");
+  });
+
+  it("maps state 3 (REM) → REM", () => {
+    expect(mapWithingsSleepState(3)).toBe("REM");
+  });
+
+  it("returns null for state 4 (synthetic marker — ignored)", () => {
+    expect(mapWithingsSleepState(4)).toBeNull();
+  });
+
+  it("returns null for any unknown state value", () => {
+    expect(mapWithingsSleepState(99)).toBeNull();
+    expect(mapWithingsSleepState(-1)).toBeNull();
+  });
+});
+
+describe("fetchWithingsSleep", () => {
+  it("POSTs sleep get with unix-seconds startdate + enddate", async () => {
+    const fetchMock = installFetchMock([]);
+    await fetchWithingsSleep("token", 1715000000, 1715100000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const [url, init] = fetchMock.mock.calls[0] as unknown as [
+      string,
+      RequestInit,
+    ];
+    expect(url).toBe("https://wbsapi.withings.net/v2/sleep");
+    const body = String(init.body);
+    expect(body).toContain("action=get");
+    expect(body).toContain("startdate=1715000000");
+    expect(body).toContain("enddate=1715100000");
+  });
+
+  it("throws when Withings returns a non-zero status", async () => {
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async () => ({
+        status: 200,
+        json: async () => ({ status: 293 }),
+      })),
+    );
+    await expect(
+      fetchWithingsSleep("token", 1715000000, 1715100000),
+    ).rejects.toThrow(/Withings sleep error: 293/);
+  });
+});
+
+describe("syncUserSleep — segment writes + idempotency", () => {
+  it("writes one row per stage segment with the mapped SleepStage", async () => {
+    // A typical night: 4 segments — light, deep, REM, light.
+    const base = 1715000000;
+    installFetchMock([
+      { startdate: base, enddate: base + 3600, state: 1, id: 99 }, // 60 min CORE
+      { startdate: base + 3600, enddate: base + 5400, state: 2, id: 99 }, // 30 min DEEP
+      { startdate: base + 5400, enddate: base + 7200, state: 3, id: 99 }, // 30 min REM
+      { startdate: base + 7200, enddate: base + 7800, state: 0, id: 99 }, // 10 min AWAKE
+    ]);
+    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
+    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+
+    const imported = await syncUserSleep("user-1");
+    expect(imported).toBe(4);
+
+    const stages = vi
+      .mocked(prisma.measurement.create)
+      .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage);
+    expect(stages).toEqual(["CORE", "DEEP", "REM", "AWAKE"]);
+
+    // First segment: 60 minutes (3600s / 60).
+    const firstDuration = vi
+      .mocked(prisma.measurement.create)
+      .mock.calls[0][0] as { data: { value: number; unit: string } };
+    expect(firstDuration.data.value).toBe(60);
+    expect(firstDuration.data.unit).toBe("minutes");
+  });
+
+  it("converts startdate (unix seconds) to a Date in measuredAt", async () => {
+    installFetchMock([
+      { startdate: 1715000000, enddate: 1715003600, state: 2, id: 1 },
+    ]);
+    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
+    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+
+    await syncUserSleep("user-1");
+    const arg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
+      data: { measuredAt: Date };
+    };
+    expect(arg.data.measuredAt.getTime()).toBe(1715000000 * 1000);
+  });
+
+  it("skips state 4 (synthetic marker) without throwing", async () => {
+    installFetchMock([
+      { startdate: 1715000000, enddate: 1715003600, state: 4, id: 1 },
+      { startdate: 1715003600, enddate: 1715007200, state: 2, id: 1 },
+    ]);
+    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
+    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+
+    const imported = await syncUserSleep("user-1");
+    expect(imported).toBe(1);
+    const stages = vi
+      .mocked(prisma.measurement.create)
+      .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage);
+    expect(stages).toEqual(["DEEP"]);
+  });
+
+  it("updates existing rows on a re-sync rather than inserting duplicates", async () => {
+    installFetchMock([
+      { startdate: 1715000000, enddate: 1715003600, state: 2, id: 1 },
+    ]);
+    // findFirst returns an existing row → writer takes the update path.
+    vi.mocked(prisma.measurement.findFirst).mockResolvedValue({
+      id: "row-1",
+    } as never);
+    vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
+
+    const imported = await syncUserSleep("user-1");
+    expect(imported).toBe(1);
+    expect(prisma.measurement.create).not.toHaveBeenCalled();
+    expect(prisma.measurement.update).toHaveBeenCalledWith({
+      where: { id: "row-1" },
+      data: { value: 60 },
+    });
+  });
+
+  it("stamps every row with an externalId tied to the segment", async () => {
+    installFetchMock([
+      { startdate: 1715000000, enddate: 1715003600, state: 2, id: 42 },
+    ]);
+    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
+    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+
+    await syncUserSleep("user-1");
+    const arg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
+      data: { externalId: string };
+    };
+    expect(arg.data.externalId).toBe("withings:sleep:user-1:42:0");
+  });
+
+  it("calls recordSyncSuccess after a clean round-trip", async () => {
+    installFetchMock([]);
+    await syncUserSleep("user-1");
+    expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
+  });
+
+  it("tolerates a ScanWatch night with no REM segment (no all-zeros row)", async () => {
+    // ScanWatch reports CORE + DEEP only; a missing REM should NOT
+    // synthesise an all-zeros REM row. Withings simply omits the
+    // segment, so the writer never sees it.
+    installFetchMock([
+      { startdate: 1715000000, enddate: 1715003600, state: 1, id: 7 },
+      { startdate: 1715003600, enddate: 1715005400, state: 2, id: 7 },
+    ]);
+    vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
+    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+
+    const imported = await syncUserSleep("user-1");
+    expect(imported).toBe(2);
+    const stages = vi
+      .mocked(prisma.measurement.create)
+      .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage);
+    expect(stages).not.toContain("REM");
+  });
+});

src/lib/withings/sync-sleep.ts

@@ -0,0 +1,250 @@
+/**
+ * v1.4.25 W17c — Withings Sleep v2 sync.
+ *
+ * Pulls per-stage sleep segments from `POST /v2/sleep?action=get` and
+ * writes one Measurement row per segment, tagged with the appropriate
+ * `SleepStage` enum. The webhook trigger (appli=44) drives most calls
+ * in production; the hourly pg-boss cron (`withings-sleep-sync`) is
+ * the safety net for the 1 % of webhook deliveries Withings drops.
+ *
+ * State mapping (research §2 + withings-go reference):
+ *
+ *   - state 0 (awake)       → AWAKE
+ *   - state 1 (light sleep) → CORE
+ *   - state 2 (deep sleep)  → DEEP
+ *   - state 3 (REM)         → REM
+ *   - state 4 (manual)      → ignored (rare; Withings synthetic marker)
+ *
+ * The legacy `ASLEEP` and `IN_BED` slots in the HealthLog enum are
+ * reserved for HealthKit ingest (pre-iOS-16 `asleepUnspecified` and
+ * iOS-16+ `inBed` respectively) and are NOT emitted from this path.
+ *
+ * Idempotency: every segment writes through the v1.4.25 W17b/c
+ * composite (Migration 0055 — `(userId, type, measuredAt, source,
+ * sleepStage)` with NULLS NOT DISTINCT). The five-tuple guarantees a
+ * re-sync upserts the same row rather than inserting a duplicate, even
+ * when Withings re-emits the night with adjusted segment boundaries.
+ *
+ * Date semantics: each segment's `measuredAt` is the segment's
+ * `startdate` (unix seconds → `Date`). The duration in minutes lands
+ * in `value`. The HealthLog analytics aggregator already groups
+ * stage rows under their parent night via the per-night `dayKey`
+ * helper, so the segment-level timestamp is the canonical sort key.
+ */
+import type { MeasurementType, SleepStage } from "@/generated/prisma/client";
+
+import { prisma } from "@/lib/db";
+import { getEvent } from "@/lib/logging/context";
+import { getUnitForType } from "@/lib/validations/measurement";
+
+import {
+  extractWithingsStatus,
+  isWithingsRefreshReauthFailure,
+} from "./sync";
+import { getValidToken } from "./sync";
+import {
+  isReauthRequired,
+  recordSyncFailure,
+  recordSyncSuccess,
+} from "@/lib/integrations/status";
+
+const WITHINGS_SLEEP_URL = "https://wbsapi.withings.net/v2/sleep";
+
+/**
+ * Default backfill window for the first sleep sync of a connection.
+ * Matches every other Withings ingest path so a reconnect lights up
+ * 30 days of stage data.
+ */
+const SLEEP_BACKFILL_DAYS = 30;
+
+/**
+ * One stage segment from the `action=get` series. Withings emits
+ * `state` as an integer enum; `startdate` and `enddate` are unix
+ * seconds (NOT YYYY-MM-DD like the activity response).
+ */
+export interface WithingsSleepSegment {
+  startdate: number;
+  enddate: number;
+  state: number;
+  // The session id is unique per night; HealthLog uses it as part of
+  // the externalId so a future re-sync can reconcile against the
+  // original write.
+  id?: number;
+}
+
+/**
+ * Map Withings `state` to the HealthLog `SleepStage` enum. Returns
+ * `null` for states we deliberately ignore (state 4 = synthetic
+ * marker) so the caller can skip the segment without an error.
+ */
+export function mapWithingsSleepState(state: number): SleepStage | null {
+  switch (state) {
+    case 0:
+      return "AWAKE";
+    case 1:
+      return "CORE";
+    case 2:
+      return "DEEP";
+    case 3:
+      return "REM";
+    default:
+      return null;
+  }
+}
+
+const SLEEP_TYPE: MeasurementType = "SLEEP_DURATION";
+
+/**
+ * Fetch the per-segment sleep series for the given user. Returns the
+ * raw segments (no DB write) so callers compose the write step.
+ */
+export async function fetchWithingsSleep(
+  accessToken: string,
+  startdate: number,
+  enddate: number,
+): Promise<WithingsSleepSegment[]> {
+  const params = new URLSearchParams({
+    action: "get",
+    startdate: String(startdate),
+    enddate: String(enddate),
+    // hr / rr / snoring are noisy intraday streams; HealthLog only
+    // ingests stage-level data today, so request the minimum set.
+  });
+
+  const pageStart = performance.now();
+  const res = await fetch(WITHINGS_SLEEP_URL, {
+    method: "POST",
+    headers: {
+      "Content-Type": "application/x-www-form-urlencoded",
+      Authorization: `Bearer ${accessToken}`,
+    },
+    body: params.toString(),
+  });
+  const json = await res.json();
+  getEvent()?.addExternalCall({
+    service: "withings",
+    method: "fetchWithingsSleep",
+    duration_ms: Math.round(performance.now() - pageStart),
+    status: res.status,
+    error: json.status !== 0 ? `status=${json.status}` : undefined,
+  });
+  if (json.status !== 0) {
+    throw new Error(`Withings sleep error: ${json.status}`);
+  }
+  const series: WithingsSleepSegment[] = json.body?.series ?? [];
+  return series;
+}
+
+/**
+ * Sync sleep data for a single user. Walks the trailing 30-day window
+ * (or whatever `SLEEP_BACKFILL_DAYS` is set to), writing one
+ * Measurement row per stage segment. Returns the number of upserted
+ * rows.
+ *
+ * Park behaviour mirrors `syncUserMeasurements` — a connection at
+ * `error_reauth` short-circuits before the upstream call.
+ */
+export async function syncUserSleep(
+  userId: string,
+  opts: { fullSync?: boolean } = {},
+): Promise<number> {
+  if (await isReauthRequired(userId, "withings")) {
+    getEvent()?.addWarning(
+      `withings sleep sync skipped for ${userId}: parked at error_reauth`,
+    );
+    return 0;
+  }
+
+  const tokenInfo = await getValidToken(userId);
+  if (!tokenInfo) return 0;
+
+  void opts.fullSync;
+  const now = new Date();
+  const start = new Date(
+    now.getTime() - SLEEP_BACKFILL_DAYS * 24 * 60 * 60 * 1000,
+  );
+  const startUnix = Math.floor(start.getTime() / 1000);
+  const endUnix = Math.floor(now.getTime() / 1000);
+
+  let segments: WithingsSleepSegment[];
+  try {
+    segments = await fetchWithingsSleep(
+      tokenInfo.accessToken,
+      startUnix,
+      endUnix,
+    );
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    await recordSyncFailure({
+      userId,
+      integration: "withings",
+      kind: isWithingsRefreshReauthFailure(message)
+        ? "reauth_required"
+        : "transient",
+      message,
+      errorCode: extractWithingsStatus(message),
+    });
+    throw err;
+  }
+
+  let imported = 0;
+  let segmentIndex = 0;
+  for (const segment of segments) {
+    const stage = mapWithingsSleepState(segment.state);
+    if (!stage) {
+      // State 4 (manual / synthetic marker) is ignored — see file
+      // header.
+      segmentIndex++;
+      continue;
+    }
+    const measuredAt = new Date(segment.startdate * 1000);
+    const durationSec = Math.max(0, segment.enddate - segment.startdate);
+    // Canonical SLEEP_DURATION unit is minutes (Apple Health alignment,
+    // v1.4.23 schema note). Round so re-sync doesn't flip values on
+    // floating-point noise.
+    const minutes = Math.round(durationSec / 60);
+
+    const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segmentIndex}`;
+
+    try {
+      const existing = await prisma.measurement.findFirst({
+        where: {
+          userId,
+          type: SLEEP_TYPE,
+          measuredAt,
+          source: "WITHINGS",
+          sleepStage: stage,
+        },
+        select: { id: true },
+      });
+      if (existing) {
+        await prisma.measurement.update({
+          where: { id: existing.id },
+          data: { value: minutes },
+        });
+      } else {
+        await prisma.measurement.create({
+          data: {
+            userId,
+            type: SLEEP_TYPE,
+            value: minutes,
+            unit: getUnitForType(SLEEP_TYPE),
+            measuredAt,
+            source: "WITHINGS",
+            sleepStage: stage,
+            externalId,
+          },
+        });
+      }
+      imported++;
+    } catch (err) {
+      getEvent()?.addWarning(
+        `Failed to upsert sleep segment (${stage}, ${measuredAt.toISOString()}): ${err}`,
+      );
+    }
+    segmentIndex++;
+  }
+
+  await recordSyncSuccess(userId, "withings");
+  return imported;
+}

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

@@ -0,0 +1,123 @@
+/**
+ * v1.4.25 W17c — Withings Sleep v2 end-to-end integration.
+ *
+ * Drives `syncUserSleep` against a mocked Withings response and the
+ * real Postgres testcontainer. Asserts the contract Migration 0055
+ * was written for: per-stage rows for the SAME night must all persist
+ * (one per stage), and a re-sync must update in place rather than
+ * duplicate them.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import { encrypt } from "@/lib/crypto";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+const TEST_USER_ID = "user-withings-sleep-sync";
+
+beforeEach(async () => {
+  const prisma = getPrismaClient();
+  await truncateAllTables(prisma);
+  await prisma.user.create({
+    data: {
+      id: TEST_USER_ID,
+      username: "withings-sleep-sync",
+      email: "withings-sleep-sync@example.test",
+    },
+  });
+  await prisma.withingsConnection.create({
+    data: {
+      userId: TEST_USER_ID,
+      withingsUserId: "wu-1",
+      accessToken: encrypt("access-token"),
+      refreshToken: encrypt("refresh-token"),
+      tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
+      scope: "user.metrics,user.activity",
+    },
+  });
+});
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+  vi.restoreAllMocks();
+});
+
+function stubWithingsSleep(series: unknown[]) {
+  vi.stubGlobal(
+    "fetch",
+    vi.fn(async () => ({
+      status: 200,
+      json: async () => ({
+        status: 0,
+        body: { series },
+      }),
+    })),
+  );
+}
+
+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 },
+      { startdate: base + 5400, enddate: base + 7200, state: 3, id: 99 },
+      { startdate: base + 7200, enddate: base + 7800, state: 0, id: 99 },
+    ]);
+
+    const { syncUserSleep } = await import("@/lib/withings/sync-sleep");
+    const imported = await syncUserSleep(TEST_USER_ID);
+    expect(imported).toBe(4);
+
+    const prisma = getPrismaClient();
+    const rows = await prisma.measurement.findMany({
+      where: { userId: TEST_USER_ID, type: "SLEEP_DURATION" },
+      orderBy: { measuredAt: "asc" },
+    });
+    expect(rows).toHaveLength(4);
+
+    // Every stage represented exactly once.
+    const stages = rows.map((r) => r.sleepStage).sort();
+    expect(stages).toEqual(["AWAKE", "CORE", "DEEP", "REM"]);
+
+    // Stage minutes — sanity check that seconds→minutes conversion is correct.
+    const byStage = new Map(rows.map((r) => [r.sleepStage, r.value]));
+    expect(byStage.get("CORE")).toBe(60);
+    expect(byStage.get("DEEP")).toBe(30);
+    expect(byStage.get("REM")).toBe(30);
+    expect(byStage.get("AWAKE")).toBe(10);
+
+    // Every row tagged as WITHINGS source — picker will treat them as
+    // a single source for the picker pipeline.
+    expect(rows.every((r) => r.source === "WITHINGS")).toBe(true);
+  });
+
+  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 },
+    ]);
+    const { syncUserSleep } = await import("@/lib/withings/sync-sleep");
+    await syncUserSleep(TEST_USER_ID);
+
+    // Withings re-aggregates the night and the DEEP segment grew to
+    // 45 minutes — second sync must UPDATE rather than INSERT.
+    stubWithingsSleep([
+      { startdate: base, enddate: base + 2700, state: 2, id: 100 },
+    ]);
+    await syncUserSleep(TEST_USER_ID);
+
+    const prisma = getPrismaClient();
+    const rows = await prisma.measurement.findMany({
+      where: { userId: TEST_USER_ID, type: "SLEEP_DURATION" },
+    });
+    expect(rows).toHaveLength(1);
+    expect(rows[0].value).toBe(45);
+    expect(rows[0].sleepStage).toBe("DEEP");
+  });
+});