From 128ee10e02daeda6016f6c34eead0ec614e32491 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 14 May 2026 18:22:58 +0200 Subject: [PATCH] =?UTF-8?q?feat(withings):=20activity=20sync=20=E2=80=94?= =?UTF-8?q?=20steps=20+=20distance=20+=20active-energy=20from=20getactivit?= =?UTF-8?q?y?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New `syncUserActivity` routine pulls daily aggregates from POST /v2/measure?action=getactivity and writes one Measurement row per (date, metric) for three fields: steps → ACTIVITY_STEPS (count) distance → WALKING_RUNNING_DISTANCE (metres) calories → ACTIVE_ENERGY_BURNED (kcal) `elevation`, soft/moderate/intense/active duration and `totalcalories` are deliberately skipped — no clean MeasurementType counterpart today; revisit when v1.5 ships the iOS workout passthrough. Rows anchor measuredAt at the day's 23:59:59 UTC end-of-day so the Berlin-timezone analytics aggregator buckets them into the correct local day even across DST shifts. Idempotency rides on the W17b composite (Migration 0055, NULLS NOT DISTINCT) — repeat syncs of the same date update value in place rather than inserting duplicates. Park behaviour mirrors syncUserMeasurements: connections at error_reauth short-circuit without making the upstream call. --- .../withings/__tests__/sync-activity.test.ts | 236 +++++++++++++++ src/lib/withings/sync-activity.ts | 284 ++++++++++++++++++ .../withings-activity-sync.test.ts | 115 +++++++ 3 files changed, 635 insertions(+) create mode 100644 src/lib/withings/__tests__/sync-activity.test.ts create mode 100644 src/lib/withings/sync-activity.ts create mode 100644 tests/integration/withings-activity-sync.test.ts diff --git a/src/lib/withings/__tests__/sync-activity.test.ts b/src/lib/withings/__tests__/sync-activity.test.ts new file mode 100644 index 000000000..52955eb38 --- /dev/null +++ b/src/lib/withings/__tests__/sync-activity.test.ts @@ -0,0 +1,236 @@ +/** + * v1.4.25 W17b — Withings activity-sync unit tests. + * + * Coverage focuses on the per-page response shape, field mapping, and + * the writer's idempotency hook. The end-to-end DB path is covered by + * 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("../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 { fetchWithingsActivity, syncUserActivity } from "../sync-activity"; + +interface FakeActivity { + date: string; + steps?: number; + distance?: number; + calories?: number; +} + +function installFetchMock(entries: FakeActivity[]) { + const fetchMock = vi.fn(async () => ({ + status: 200, + json: async () => ({ + status: 0, + body: { + activities: entries, + more: false, + offset: 0, + }, + }), + })); + vi.stubGlobal("fetch", fetchMock); + return fetchMock; +} + +beforeEach(() => { + vi.resetAllMocks(); +}); + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("fetchWithingsActivity", () => { + it("POSTs getactivity with steps + distance + calories in data_fields", async () => { + const fetchMock = installFetchMock([]); + await fetchWithingsActivity("token", "2026-05-01", "2026-05-12"); + + expect(fetchMock).toHaveBeenCalledTimes(1); + const [url, init] = fetchMock.mock.calls[0] as unknown as [ + string, + RequestInit, + ]; + expect(url).toBe("https://wbsapi.withings.net/v2/measure"); + const body = String(init.body); + expect(body).toContain("action=getactivity"); + expect(body).toContain("startdateymd=2026-05-01"); + expect(body).toContain("enddateymd=2026-05-12"); + expect(body).toContain("data_fields=steps%2Cdistance%2Ccalories"); + }); + + it("returns one entry per day from the response", async () => { + installFetchMock([ + { date: "2026-05-10", steps: 8420, distance: 6720, calories: 412 }, + { date: "2026-05-11", steps: 5012, distance: 3950, calories: 245 }, + ]); + const entries = await fetchWithingsActivity( + "token", + "2026-05-10", + "2026-05-11", + ); + expect(entries).toHaveLength(2); + expect(entries[0].steps).toBe(8420); + expect(entries[1].date).toBe("2026-05-11"); + }); + + it("throws when Withings returns a non-zero status", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async () => ({ + status: 200, + json: async () => ({ status: 293, error: "insufficient scope" }), + })), + ); + await expect( + fetchWithingsActivity("token", "2026-05-10", "2026-05-11"), + ).rejects.toThrow(/Withings activity error: 293/); + }); +}); + +describe("syncUserActivity — field mapping + idempotency", () => { + it("writes one row per (date, metric) the first time a day shows up", async () => { + installFetchMock([ + { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 }, + ]); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + const imported = await syncUserActivity("user-1"); + + expect(imported).toBe(3); + expect(prisma.measurement.create).toHaveBeenCalledTimes(3); + + const types = vi + .mocked(prisma.measurement.create) + .mock.calls.map((c) => (c[0].data as { type: string }).type); + expect(types.sort()).toEqual([ + "ACTIVE_ENERGY_BURNED", + "ACTIVITY_STEPS", + "WALKING_RUNNING_DISTANCE", + ]); + }); + + it("updates instead of inserting when the same (date, metric) already exists", async () => { + installFetchMock([ + { date: "2026-05-12", steps: 9001, distance: 7000, calories: 450 }, + ]); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue({ + id: "row-existing", + } as never); + vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); + + const imported = await syncUserActivity("user-1"); + + expect(imported).toBe(3); + expect(prisma.measurement.create).not.toHaveBeenCalled(); + expect(prisma.measurement.update).toHaveBeenCalledTimes(3); + }); + + it("anchors measuredAt at the day's 23:59:59 UTC end-of-day", async () => { + installFetchMock([{ date: "2026-05-12", steps: 1000 }]); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + await syncUserActivity("user-1"); + + const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as { + data: { measuredAt: Date }; + }; + const measuredAt = createArg.data.measuredAt; + expect(measuredAt.toISOString()).toBe("2026-05-12T23:59:59.000Z"); + }); + + it("skips missing fields without writing zero rows or throwing", async () => { + // `distance` absent — Withings drops the field rather than emitting 0 + // when there's no GPS. Steps + calories still write. + installFetchMock([{ date: "2026-05-12", steps: 8000, calories: 380 }]); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + const imported = await syncUserActivity("user-1"); + + expect(imported).toBe(2); + const types = vi + .mocked(prisma.measurement.create) + .mock.calls.map((c) => (c[0].data as { type: string }).type); + expect(types).not.toContain("WALKING_RUNNING_DISTANCE"); + }); + + it("ingests a 0-step rest day rather than dropping it", async () => { + // Withings reports 0 steps for a day a user spent in bed — that's + // a valid datapoint, not a missing one. The mapper must not skip + // it (which would let the prior day's value erroneously "stick"). + installFetchMock([ + { date: "2026-05-12", steps: 0, distance: 0, calories: 0 }, + ]); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + const imported = await syncUserActivity("user-1"); + expect(imported).toBe(3); + }); + + it("calls recordSyncSuccess after a clean round-trip", async () => { + installFetchMock([{ date: "2026-05-12", steps: 1000 }]); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + await syncUserActivity("user-1"); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + }); + + it("tags every row with the canonical externalId so future replays dedup", async () => { + installFetchMock([{ date: "2026-05-12", steps: 8420 }]); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + await syncUserActivity("user-1"); + + const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as { + data: { externalId: string }; + }; + expect(createArg.data.externalId).toBe( + "withings:activity:user-1:2026-05-12:steps", + ); + }); +}); diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts new file mode 100644 index 000000000..591404eab --- /dev/null +++ b/src/lib/withings/sync-activity.ts @@ -0,0 +1,284 @@ +/** + * v1.4.25 W17b — Withings Activity sync. + * + * Pulls daily aggregates from `POST /v2/measure?action=getactivity` + * and writes one Measurement row per (date, metric). The webhook + * trigger (appli=16) drives most calls in production; the hourly + * pg-boss cron (`withings-activity-sync`) is the safety net for the + * 1 % of webhook deliveries Withings drops. + * + * Field mapping (research §1 + §4): + * + * - `steps` → ACTIVITY_STEPS (count) + * - `distance` → WALKING_RUNNING_DISTANCE (metres) + * - `calories` → ACTIVE_ENERGY_BURNED (kcal) + * + * `elevation`, `totalcalories`, soft/moderate/intense/active duration + * are deliberately skipped — no clean MeasurementType counterpart + * today. Re-evaluated when v1.5 ships the iOS workout passthrough. + * + * Idempotency: every row writes through the v1.4.25 W17b/c composite + * unique (Migration 0055 — `(userId, type, measuredAt, source, + * sleepStage)` with NULLS NOT DISTINCT). `sleepStage` is NULL for + * every activity row, so the four-column dedup behaviour is unchanged + * and the same date re-syncing simply updates `value`. + * + * Date semantics: Withings reports per-day aggregates with `date` as + * `YYYY-MM-DD`. We anchor `measuredAt` at the day's 23:59:59 UTC so + * the row sorts at the end of the day in chronological reads and the + * Berlin-day analytics aggregator (timezone-aware) buckets it into the + * correct day even when DST shifts the boundary. + */ +import type { MeasurementType } 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_MEASURE_URL = "https://wbsapi.withings.net/v2/measure"; + +/** + * Default backfill window for the FIRST activity sync of a connection. + * Matches the measurement-sync default in `sync.ts` so a freshly + * reconnected user lights up the same 30 days of history across every + * Withings ingest path. + */ +const ACTIVITY_BACKFILL_DAYS = 30; + +/** + * Withings `getactivity` response shape — one entry per calendar day. + * Only the fields we consume are typed; the response carries 15+ more + * fields that future waves may pull in. + * + * https://developer.withings.com/api-reference#tag/measure/operation/measurev2-getactivity + */ +export interface WithingsActivityEntry { + date: string; // YYYY-MM-DD + timezone?: string; + steps?: number; + distance?: number; // metres + calories?: number; // active kcal (NOT totalcalories) + // The fields below are not ingested today (see file header): + elevation?: number; + totalcalories?: number; + soft?: number; + moderate?: number; + intense?: number; + active?: number; +} + +/** + * Per-row mapping table. Each entry tells the writer: which field on + * the activity entry to read, which MeasurementType bucket to write, + * and which canonical unit string to stamp on the row. Adding a new + * field (e.g. `elevation` once FLIGHTS_CLIMBED canonicalisation + * settles) is one row here plus no other code change. + */ +const ACTIVITY_FIELD_MAP: ReadonlyArray<{ + field: keyof Pick; + type: MeasurementType; +}> = [ + { field: "steps", type: "ACTIVITY_STEPS" }, + { field: "distance", type: "WALKING_RUNNING_DISTANCE" }, + { field: "calories", type: "ACTIVE_ENERGY_BURNED" }, +]; + +/** + * Anchor a per-day activity row at the day's last second UTC. Berlin- + * timezone day analytics consume the row through the standard + * `dayKey()` helper, which buckets it into the right local day even + * across DST shifts. + */ +function activityMeasuredAt(yyyymmdd: string): Date { + // The Withings response is always `YYYY-MM-DD` — Date parses the ISO + // form with implicit UTC midnight. Add 23h 59m 59s so the row sits + // at end-of-day; the analytics layer day-keys it the same way it + // does every other end-of-day rollup. + const base = new Date(`${yyyymmdd}T23:59:59.000Z`); + return base; +} + +/** + * Fetch activity entries from Withings for the given user. Returns + * the raw entries (no DB write) so callers can compose the write + * step (which itself uses Prisma helpers). + */ +export async function fetchWithingsActivity( + accessToken: string, + startdateymd: string, + enddateymd: string, +): Promise { + const baseParams: Record = { + action: "getactivity", + startdateymd, + enddateymd, + // The Withings getactivity response carries optional fields only + // when explicitly listed in `data_fields`. Steps + distance + + // calories cover the W17b mapping; future expansion adds fields + // here without touching the write path. + data_fields: "steps,distance,calories", + }; + + const results: WithingsActivityEntry[] = []; + let offset = 0; + let pageCount = 0; + while (true) { + const params = new URLSearchParams({ + ...baseParams, + offset: String(offset), + }); + const pageStart = performance.now(); + const res = await fetch(WITHINGS_MEASURE_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: `fetchWithingsActivity(page=${pageCount})`, + 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 activity error: ${json.status}`); + } + + const body = json.body ?? {}; + const activities: WithingsActivityEntry[] = body.activities ?? []; + for (const entry of activities) { + results.push(entry); + } + + const hasMore = body.more === true || body.more === 1; + if (!hasMore) break; + const nextOffset = Number(body.offset); + if (!Number.isFinite(nextOffset) || nextOffset <= offset) break; + offset = nextOffset; + pageCount += 1; + if (pageCount > 1000) break; + } + return results; +} + +/** + * Sync activity data for a single user. Writes one Measurement row + * per (date, metric) where the corresponding Withings field is + * present and finite. Returns the number of upserted rows. + * + * Parks at `error_reauth` short-circuit identical to + * `syncUserMeasurements` — the user has to redo OAuth before the next + * cron tick will attempt anything. + */ +export async function syncUserActivity( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + if (await isReauthRequired(userId, "withings")) { + getEvent()?.addWarning( + `withings activity sync skipped for ${userId}: parked at error_reauth`, + ); + return 0; + } + + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + // 30-day rolling window keeps the call cheap (≤ one page) and + // self-heals stale rows from a clock skew or a backfill replay. + // The webhook-driven path narrows naturally to today/yesterday; + // the cron fallback walks the full window. + void opts.fullSync; // reserved for future expansion + const now = new Date(); + const start = new Date( + now.getTime() - ACTIVITY_BACKFILL_DAYS * 24 * 60 * 60 * 1000, + ); + const ymd = (d: Date) => d.toISOString().slice(0, 10); + + let entries: WithingsActivityEntry[]; + try { + entries = await fetchWithingsActivity( + tokenInfo.accessToken, + ymd(start), + ymd(now), + ); + } 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; + for (const entry of entries) { + if (!entry.date) continue; + const measuredAt = activityMeasuredAt(entry.date); + for (const { field, type } of ACTIVITY_FIELD_MAP) { + const raw = entry[field]; + // Withings returns 0 for "active but no movement" — that's + // valid (a day of rest still records 0 steps). Only skip + // undefined / null / NaN. + if (raw == null || !Number.isFinite(raw)) continue; + try { + const existing = await prisma.measurement.findFirst({ + where: { + userId, + type, + measuredAt, + source: "WITHINGS", + sleepStage: null, + }, + select: { id: true }, + }); + if (existing) { + await prisma.measurement.update({ + where: { id: existing.id }, + data: { value: raw }, + }); + } else { + await prisma.measurement.create({ + data: { + userId, + type, + value: raw, + unit: getUnitForType(type), + measuredAt, + source: "WITHINGS", + externalId: `withings:activity:${userId}:${entry.date}:${field}`, + }, + }); + } + imported++; + } catch (err) { + getEvent()?.addWarning( + `Failed to upsert activity row (${type}, ${entry.date}): ${err}`, + ); + } + } + } + + await recordSyncSuccess(userId, "withings"); + return imported; +} diff --git a/tests/integration/withings-activity-sync.test.ts b/tests/integration/withings-activity-sync.test.ts new file mode 100644 index 000000000..5afc464b2 --- /dev/null +++ b/tests/integration/withings-activity-sync.test.ts @@ -0,0 +1,115 @@ +/** + * v1.4.25 W17b — Withings activity-sync end-to-end. + * + * Drives `syncUserActivity` against a mocked Withings response and the + * real Postgres testcontainer. Asserts: + * - Per-day activity entries map to one row per (date, metric) + * - The composite-unique with NULLS NOT DISTINCT serializes a + * second sync of the same window into UPDATEs (no duplicates) + * - Activity rows leave sleep_stage NULL so they do not collide + * with the W17c sleep rows + */ +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-activity-sync"; + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: TEST_USER_ID, + username: "withings-activity-sync", + email: "withings-activity-sync@example.test", + }, + }); + await prisma.withingsConnection.create({ + data: { + userId: TEST_USER_ID, + withingsUserId: "wu-1", + accessToken: encrypt("access-token"), + refreshToken: encrypt("refresh-token"), + // Token valid for an hour — keeps `getValidToken` off the + // refresh path. + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + scope: "user.metrics,user.activity", + }, + }); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); +}); + +function stubWithingsActivity(entries: unknown[]) { + vi.stubGlobal( + "fetch", + vi.fn(async () => ({ + status: 200, + json: async () => ({ + status: 0, + body: { activities: entries, more: false, offset: 0 }, + }), + })), + ); +} + +describe("syncUserActivity — integration", () => { + it("writes one Measurement row per (date, metric) on first sync", async () => { + stubWithingsActivity([ + { date: "2026-05-10", steps: 8420, distance: 6720, calories: 412 }, + { date: "2026-05-11", steps: 5012, distance: 3950, calories: 245 }, + ]); + + const { syncUserActivity } = await import("@/lib/withings/sync-activity"); + const imported = await syncUserActivity(TEST_USER_ID); + expect(imported).toBe(6); // 2 days × 3 fields + + const prisma = getPrismaClient(); + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, source: "WITHINGS" }, + orderBy: [{ measuredAt: "asc" }, { type: "asc" }], + }); + expect(rows).toHaveLength(6); + // Activity rows never carry a sleepStage — confirms the new + // composite leaves the column NULL for non-sleep ingest. + expect(rows.every((r) => r.sleepStage === null)).toBe(true); + + const types = new Set(rows.map((r) => r.type)); + expect(types.has("ACTIVITY_STEPS")).toBe(true); + expect(types.has("WALKING_RUNNING_DISTANCE")).toBe(true); + expect(types.has("ACTIVE_ENERGY_BURNED")).toBe(true); + }); + + it("re-running the same sync updates rather than duplicating", async () => { + const entries = [ + { date: "2026-05-10", steps: 8420, distance: 6720, calories: 412 }, + ]; + stubWithingsActivity(entries); + const { syncUserActivity } = await import("@/lib/withings/sync-activity"); + await syncUserActivity(TEST_USER_ID); + + // Withings re-aggregates a day when late samples arrive; second + // pull returns updated values. + stubWithingsActivity([ + { date: "2026-05-10", steps: 9001, distance: 7200, calories: 450 }, + ]); + await syncUserActivity(TEST_USER_ID); + + const prisma = getPrismaClient(); + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, source: "WITHINGS" }, + orderBy: { type: "asc" }, + }); + expect(rows).toHaveLength(3); + const steps = rows.find((r) => r.type === "ACTIVITY_STEPS"); + expect(steps?.value).toBe(9001); + const distance = rows.find((r) => r.type === "WALKING_RUNNING_DISTANCE"); + expect(distance?.value).toBe(7200); + }); +});