feat(withings): activity sync — steps + distance + active-energy from getactivity

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.
This commit is contained in:
Marc-André Bombeck committed 2026-05-14 18:22:58 +02:00
1 parent 0af9923b46
commit 128ee10e02
3 files changed
+635

No files matched your search

@@ -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<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 { 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",
);
});
});
+284
View File
@@ -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<WithingsActivityEntry, "steps" | "distance" | "calories">;
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<WithingsActivityEntry[]> {
const baseParams: Record<string, string> = {
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<number> {
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;
}
@@ -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);
});
});