diff --git a/src/lib/fitbit/__tests__/client.test.ts b/src/lib/fitbit/__tests__/client.test.ts index 7410c17ac..392c818ab 100644 --- a/src/lib/fitbit/__tests__/client.test.ts +++ b/src/lib/fitbit/__tests__/client.test.ts @@ -398,6 +398,56 @@ describe("VO2 max", () => { }); }); +describe("mappers in the user's zone", () => { + it("reads a weigh-in's logged wall clock in the user's zone", () => { + const entry = { + weight: [ + { date: "2026-05-10", time: "07:30:00", weight: 81.4, logId: 7 }, + ], + }; + expect( + mapWeight(entry, "Pacific/Auckland")[0]!.measuredAt.toISOString(), + ).toBe("2026-05-09T19:30:00.000Z"); + expect( + mapWeight(entry, "America/New_York")[0]!.measuredAt.toISOString(), + ).toBe("2026-05-10T11:30:00.000Z"); + // The externalId does not depend on the zone. + expect(mapWeight(entry, "Pacific/Auckland")[0]!.fieldTag).toBe("7:weight"); + }); + + it("keeps a logId-less weigh-in's externalId as earlier versions spelled it", () => { + const out = mapWeight( + { weight: [{ date: "2026-05-10", time: "07:30:00", weight: 81.4 }] }, + "Pacific/Auckland", + ); + expect(out[0]!.fieldTag).toBe("2026-05-10T07:30:00.000Z:weight"); + }); + + it.each([ + ["Pacific/Tongatapu", "2026-05-09T23:00:00.000Z"], + ["Pacific/Auckland", "2026-05-10T00:00:00.000Z"], + ["America/Santiago", "2026-05-10T16:00:00.000Z"], + ["America/New_York", "2026-05-10T16:00:00.000Z"], + ["Europe/Berlin", "2026-05-10T10:00:00.000Z"], + ])( + "anchors a daily total at local noon in %s, on the same day there", + (tz, iso) => { + const [steps] = mapSteps( + { "activities-steps": [{ dateTime: "2026-05-10", value: "8421" }] }, + tz, + ); + expect(steps!.measuredAt.toISOString()).toBe(iso); + expect(steps!.fieldTag).toBe("steps:2026-05-10"); + const [spo2] = mapOxygenSaturation( + [{ dateTime: "2026-05-10", value: { avg: 97 } }], + tz, + ); + expect(spo2!.measuredAt.toISOString()).toBe(iso); + expect(spo2!.fieldTag).toBe("2026-05-10:spo2"); + }, + ); +}); + describe("activity mappers (cumulative daily)", () => { it("maps steps from the string value, preserving a zero, with a day-keyed externalId", () => { const out = mapSteps({ diff --git a/src/lib/fitbit/__tests__/sync.test.ts b/src/lib/fitbit/__tests__/sync.test.ts index 9974e0b50..9d8990a3b 100644 --- a/src/lib/fitbit/__tests__/sync.test.ts +++ b/src/lib/fitbit/__tests__/sync.test.ts @@ -288,7 +288,12 @@ describe("upsertFitbitMeasurements — batched write, tombstones resurrect", () const measuredAt = new Date("2026-05-10T07:00:00.000Z"); // A re-fetched daily summary: the live row already exists. prismaMock.measurement.findMany.mockResolvedValue([ - { id: "m-live", type: "WEIGHT", externalId: "stats:weight:2026-05-10" }, + { + id: "m-live", + type: "WEIGHT", + externalId: "stats:weight:2026-05-10", + measuredAt, + }, ]); const { imported } = await upsertFitbitMeasurements("user1", [ @@ -319,7 +324,12 @@ describe("upsertFitbitMeasurements — batched write, tombstones resurrect", () // probe returns it; planning an insert instead would be dropped silently // by `skipDuplicates` and wedge the key forever. prismaMock.measurement.findMany.mockResolvedValue([ - { id: "m-dead", type: "WEIGHT", externalId: "stats:weight:2026-05-10" }, + { + id: "m-dead", + type: "WEIGHT", + externalId: "stats:weight:2026-05-10", + measuredAt, + }, ]); const { imported } = await upsertFitbitMeasurements("user1", [ @@ -342,11 +352,52 @@ describe("upsertFitbitMeasurements — batched write, tombstones resurrect", () expect(updateArg.data.deletedAt).toBeNull(); }); + it("moves a daily row to its new anchor and refolds the day it left", async () => { + // A step total written at the old noon-UTC anchor, re-fetched for a user + // in Tonga, whose local noon is 23:00 UTC the day before. + const legacy = new Date("2026-05-10T12:00:00.000Z"); + const localNoon = new Date("2026-05-09T23:00:00.000Z"); + prismaMock.measurement.findMany.mockResolvedValue([ + { + id: "m-steps", + type: "ACTIVITY_STEPS", + externalId: "stats:steps:2026-05-10", + value: 8421, + unit: "steps", + measuredAt: legacy, + sleepStage: null, + deletedAt: null, + }, + ]); + + const { touched } = await upsertFitbitMeasurements("user1", [ + { + type: "ACTIVITY_STEPS", + value: 8421, + unit: "steps", + measuredAt: localNoon, + externalId: "stats:steps:2026-05-10", + }, + ]); + + expect(prismaMock.measurement.createManyAndReturn).not.toHaveBeenCalled(); + expect(prismaMock.measurement.update).toHaveBeenCalledTimes(1); + const updateArg = prismaMock.measurement.update.mock.calls[0]![0] as { + data: { measuredAt: Date }; + }; + expect(updateArg.data.measuredAt).toEqual(localNoon); + expect(touched.map((t) => t.measuredAt.toISOString())).toEqual( + expect.arrayContaining([localNoon.toISOString(), legacy.toISOString()]), + ); + }); + it("splits a mixed batch: matched-live → update, unmatched → createMany", async () => { const measuredAt = new Date("2026-05-10T07:00:00.000Z"); - prismaMock.measurement.findMany.mockResolvedValue([ - { id: "m-live", type: "WEIGHT", externalId: "live-key" }, - ]); + prismaMock.measurement.findMany + .mockResolvedValueOnce([ + { id: "m-live", type: "WEIGHT", externalId: "live-key", measuredAt }, + ]) + .mockResolvedValueOnce([]); // natural-key rescue: no twin const { imported } = await upsertFitbitMeasurements("user1", [ { @@ -438,7 +489,7 @@ describe("upsertFitbitMeasurements — write-catch ledger (F-2)", () => { // fresh → create branch. Second findMany is the natural-key rescue probe. prismaMock.measurement.findMany .mockResolvedValueOnce([ - { id: "m-live", type: "WEIGHT", externalId: "live-key" }, + { id: "m-live", type: "WEIGHT", externalId: "live-key", measuredAt }, ]) .mockResolvedValueOnce([]); // natural-key rescue: no twin prismaMock.measurement.createManyAndReturn.mockRejectedValueOnce( @@ -478,8 +529,8 @@ describe("upsertFitbitMeasurements — write-catch ledger (F-2)", () => { // throws; the loop must continue to the second, and the create still runs. prismaMock.measurement.findMany .mockResolvedValueOnce([ - { id: "m-live-1", type: "WEIGHT", externalId: "live-1" }, - { id: "m-live-2", type: "WEIGHT", externalId: "live-2" }, + { id: "m-live-1", type: "WEIGHT", externalId: "live-1", measuredAt }, + { id: "m-live-2", type: "WEIGHT", externalId: "live-2", measuredAt }, ]) .mockResolvedValueOnce([]); // natural-key rescue: no twin prismaMock.measurement.update diff --git a/src/lib/fitbit/client.ts b/src/lib/fitbit/client.ts index 1015e15fa..6521cd8f0 100644 --- a/src/lib/fitbit/client.ts +++ b/src/lib/fitbit/client.ts @@ -39,6 +39,7 @@ import { createHash, randomBytes } from "node:crypto"; import { getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; +import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; import { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; import { FitbitApiError, classifyFitbitResponse } from "./response-classifier"; @@ -528,15 +529,16 @@ function toFiniteNumber(v: unknown): number | null { } /** - * Parse a Fitbit `YYYY-MM-DD` civil date into a UTC-midday Date. Anchored at - * midday so a timezone shift can't roll the civil day across a boundary. Returns - * null for an unparseable string. + * Validate a Fitbit `YYYY-MM-DD` civil date and return it as a day key, or + * null for an unparseable string. The key names the row's externalId; the + * row's instant is the key's local noon in the user's zone + * (`canonicalDailyTimestamp`), which reads the same calendar day there. + * Noon UTC, the previous anchor, is already the next day from UTC+13 on. */ -function parseCivilDate(dateStr: unknown): Date | null { +function parseCivilKey(dateStr: unknown): string | null { if (typeof dateStr !== "string") return null; const m = /^(\d{4})-(\d{2})-(\d{2})$/.exec(dateStr.trim()); - if (!m) return null; - return new Date(Date.UTC(Number(m[1]), Number(m[2]) - 1, Number(m[3]), 12)); + return m ? `${m[1]}-${m[2]}-${m[3]}` : null; } /** @@ -584,28 +586,45 @@ interface FitbitBodyLogEntry { fat?: number; } -/** Resolve the measuredAt for a body-log entry from its `date` + `time`. */ -function bodyLogInstant(entry: FitbitBodyLogEntry): Date { +/** + * Resolve the measuredAt for a body-log entry from its `date` + `time`. Both + * are the user's local wall clock (Fitbit logs carry no offset), so they are + * read in the user's zone. Reading them as UTC put a morning weigh-in hours + * off, and on the previous or next day for anyone far from UTC. + */ +function bodyLogInstant(entry: FitbitBodyLogEntry, tz?: string): Date { const date = typeof entry.date === "string" ? entry.date : undefined; const time = typeof entry.time === "string" ? entry.time : "12:00:00"; if (date) { - const d = new Date(`${date}T${time}Z`); - if (!Number.isNaN(d.getTime())) return d; - const civil = parseCivilDate(date); - if (civil) return civil; + const d = parseLocalInstant(`${date}T${time}`, tz ?? "UTC"); + if (d) return d; + const civil = parseCivilKey(date); + if (civil) return canonicalDailyTimestamp(civil, tz); } return new Date(); } -/** Stable anchor for a body-log externalId — the logId, else the instant. */ +/** + * Stable anchor for a body-log externalId — the logId, else the logged wall + * clock. The wall clock is spelled the way earlier versions spelled it (as + * if it were UTC), so an entry without a logId keeps its externalId. + */ function bodyLogAnchor(entry: FitbitBodyLogEntry): string { if (typeof entry.logId === "number" && Number.isFinite(entry.logId)) { return String(entry.logId); } - return bodyLogInstant(entry).toISOString(); + const date = typeof entry.date === "string" ? entry.date : undefined; + const time = typeof entry.time === "string" ? entry.time : "12:00:00"; + const asWritten = date ? new Date(`${date}T${time}Z`) : null; + return asWritten && !Number.isNaN(asWritten.getTime()) + ? asWritten.toISOString() + : bodyLogInstant(entry).toISOString(); } -export function mapWeight(body: unknown): FitbitMappedMeasurement[] { +export function mapWeight( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { const arr = readArray(body, "weight"); const out: FitbitMappedMeasurement[] = []; for (const raw of arr) { @@ -615,14 +634,17 @@ export function mapWeight(body: unknown): FitbitMappedMeasurement[] { type: "WEIGHT", value: round2(e.weight), unit: "kg", - measuredAt: bodyLogInstant(e), + measuredAt: bodyLogInstant(e, tz), fieldTag: `${bodyLogAnchor(e)}:weight`, }); } return out; } -export function mapBodyFat(body: unknown): FitbitMappedMeasurement[] { +export function mapBodyFat( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { const arr = readArray(body, "fat"); const out: FitbitMappedMeasurement[] = []; for (const raw of arr) { @@ -632,7 +654,7 @@ export function mapBodyFat(body: unknown): FitbitMappedMeasurement[] { type: "BODY_FAT", value: round2(e.fat), unit: "%", - measuredAt: bodyLogInstant(e), + measuredAt: bodyLogInstant(e, tz), fieldTag: `${bodyLogAnchor(e)}:body_fat`, }); } @@ -665,10 +687,11 @@ function mapDailySummary( leaves: string[]; factor?: number; }, + tz: string | undefined, ): FitbitMappedMeasurement[] { const out: FitbitMappedMeasurement[] = []; for (const row of rows) { - const day = parseCivilDate(row.dateTime); + const day = parseCivilKey(row.dateTime); if (!day) continue; const value = firstPositiveLeaf(row.value, spec.leaves); if (value === null) continue; @@ -677,8 +700,8 @@ function mapDailySummary( type: spec.type, value: round2(scaled), unit: spec.unit, - measuredAt: day, - fieldTag: `${fitbitDate(day)}:${spec.fieldTag}`, + measuredAt: canonicalDailyTimestamp(day, tz), + fieldTag: `${day}:${spec.fieldTag}`, }); } return out; @@ -694,41 +717,60 @@ function firstPositiveLeaf(obj: unknown, leaves: string[]): number | null { return null; } -export function mapOxygenSaturation(body: unknown): FitbitMappedMeasurement[] { +export function mapOxygenSaturation( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { // The SpO2 summary endpoint returns a BARE ARRAY of daily rows (not wrapped). const rows = Array.isArray(body) ? (body as FitbitDailyRow[]) : []; - return mapDailySummary(rows, { - type: "OXYGEN_SATURATION", - unit: "%", - fieldTag: "spo2", - leaves: ["avg"], - }); + return mapDailySummary( + rows, + { + type: "OXYGEN_SATURATION", + unit: "%", + fieldTag: "spo2", + leaves: ["avg"], + }, + tz, + ); } export function mapHeartRateVariability( body: unknown, + tz?: string, ): FitbitMappedMeasurement[] { // Fitbit's HRV summary is an RMSSD estimator (`value.dailyRmssd`). It lands in // the canonical `HEART_RATE_VARIABILITY` slot that FITBIT occupies in the // source-priority `hrv` ladder (alongside Apple / Oura), keeping cross-source // comparison on one axis. (`HRV_RMSSD` is reserved for the WHOOP-native slot.) const rows = readArray(body, "hrv") as FitbitDailyRow[]; - return mapDailySummary(rows, { - type: "HEART_RATE_VARIABILITY", - unit: "ms", - fieldTag: "hrv", - leaves: ["dailyRmssd"], - }); + return mapDailySummary( + rows, + { + type: "HEART_RATE_VARIABILITY", + unit: "ms", + fieldTag: "hrv", + leaves: ["dailyRmssd"], + }, + tz, + ); } -export function mapRespiratoryRate(body: unknown): FitbitMappedMeasurement[] { +export function mapRespiratoryRate( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { const rows = readArray(body, "br") as FitbitDailyRow[]; - return mapDailySummary(rows, { - type: "RESPIRATORY_RATE", - unit: "breaths/min", - fieldTag: "resp_rate", - leaves: ["breathingRate"], - }); + return mapDailySummary( + rows, + { + type: "RESPIRATORY_RATE", + unit: "breaths/min", + fieldTag: "resp_rate", + leaves: ["breathingRate"], + }, + tz, + ); } /** @@ -736,14 +778,21 @@ export function mapRespiratoryRate(body: unknown): FitbitMappedMeasurement[] { * GET /1/user/-/activities/heart/date/{start}/{end}.json * → { "activities-heart": [{ dateTime, value: { restingHeartRate } }] } */ -export function mapRestingHeartRate(body: unknown): FitbitMappedMeasurement[] { +export function mapRestingHeartRate( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { const rows = readArray(body, "activities-heart") as FitbitDailyRow[]; - return mapDailySummary(rows, { - type: "RESTING_HEART_RATE", - unit: "bpm", - fieldTag: "rhr", - leaves: ["restingHeartRate"], - }); + return mapDailySummary( + rows, + { + type: "RESTING_HEART_RATE", + unit: "bpm", + fieldTag: "rhr", + leaves: ["restingHeartRate"], + }, + tz, + ); } /** @@ -753,11 +802,14 @@ export function mapRestingHeartRate(body: unknown): FitbitMappedMeasurement[] { * The value is a STRING that may be a single number or a range; a range resolves * to its midpoint. Daily latest-wins, day-keyed externalId. */ -export function mapVo2Max(body: unknown): FitbitMappedMeasurement[] { +export function mapVo2Max( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { const rows = readArray(body, "cardioScore") as FitbitDailyRow[]; const out: FitbitMappedMeasurement[] = []; for (const row of rows) { - const day = parseCivilDate(row.dateTime); + const day = parseCivilKey(row.dateTime); if (!day) continue; const v = parseVo2Max( (row.value as Record | undefined)?.vo2Max, @@ -767,8 +819,8 @@ export function mapVo2Max(body: unknown): FitbitMappedMeasurement[] { type: "VO2_MAX", value: round2(v), unit: "mL/(kg·min)", - measuredAt: day, - fieldTag: `vo2_max:${fitbitDate(day)}`, + measuredAt: canonicalDailyTimestamp(day, tz), + fieldTag: `vo2_max:${day}`, cumulativeDaily: true, }); } @@ -811,11 +863,12 @@ function mapActivitySeries( fieldTag: string; factor?: number; }, + tz: string | undefined, ): FitbitMappedMeasurement[] { const rows = readArray(body, spec.arrayKey) as FitbitDailyRow[]; const out: FitbitMappedMeasurement[] = []; for (const row of rows) { - const day = parseCivilDate(row.dateTime); + const day = parseCivilKey(row.dateTime); if (!day) continue; const n = toFiniteNumber(row.value); if (n === null || !nonNegative(n)) continue; @@ -824,52 +877,80 @@ function mapActivitySeries( type: spec.type, value: round2(scaled), unit: spec.unit, - measuredAt: day, - fieldTag: `${spec.fieldTag}:${fitbitDate(day)}`, + measuredAt: canonicalDailyTimestamp(day, tz), + fieldTag: `${spec.fieldTag}:${day}`, cumulativeDaily: true, }); } return out; } -export function mapSteps(body: unknown): FitbitMappedMeasurement[] { - return mapActivitySeries(body, { - arrayKey: "activities-steps", - type: "ACTIVITY_STEPS", - unit: "steps", - fieldTag: "steps", - }); +export function mapSteps( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { + return mapActivitySeries( + body, + { + arrayKey: "activities-steps", + type: "ACTIVITY_STEPS", + unit: "steps", + fieldTag: "steps", + }, + tz, + ); } -export function mapDistance(body: unknown): FitbitMappedMeasurement[] { +export function mapDistance( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { // The metric-locale distance series reports kilometres → metres. - return mapActivitySeries(body, { - arrayKey: "activities-distance", - type: "WALKING_RUNNING_DISTANCE", - unit: "m", - fieldTag: "distance", - factor: 1000, - }); + return mapActivitySeries( + body, + { + arrayKey: "activities-distance", + type: "WALKING_RUNNING_DISTANCE", + unit: "m", + fieldTag: "distance", + factor: 1000, + }, + tz, + ); } -export function mapActiveCalories(body: unknown): FitbitMappedMeasurement[] { +export function mapActiveCalories( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { // `activityCalories` is the ACTIVE portion (excludes BMR), unlike the total // `calories` resource. - return mapActivitySeries(body, { - arrayKey: "activities-activityCalories", - type: "ACTIVE_ENERGY_BURNED", - unit: "kcal", - fieldTag: "active_calories", - }); + return mapActivitySeries( + body, + { + arrayKey: "activities-activityCalories", + type: "ACTIVE_ENERGY_BURNED", + unit: "kcal", + fieldTag: "active_calories", + }, + tz, + ); } -export function mapFloors(body: unknown): FitbitMappedMeasurement[] { - return mapActivitySeries(body, { - arrayKey: "activities-floors", - type: "FLIGHTS_CLIMBED", - unit: "flights", - fieldTag: "floors", - }); +export function mapFloors( + body: unknown, + tz?: string, +): FitbitMappedMeasurement[] { + return mapActivitySeries( + body, + { + arrayKey: "activities-floors", + type: "FLIGHTS_CLIMBED", + unit: "flights", + fieldTag: "floors", + }, + tz, + ); } /** diff --git a/src/lib/fitbit/sync-activity.ts b/src/lib/fitbit/sync-activity.ts index a0451fd5d..10e1140ac 100644 --- a/src/lib/fitbit/sync-activity.ts +++ b/src/lib/fitbit/sync-activity.ts @@ -30,6 +30,8 @@ import { mapVo2Max, } from "./client"; import type { FitbitMappedMeasurement } from "./client"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; +import { localTodayAsDate } from "./sync-core"; import { chunkDateRanges, getValidToken, @@ -45,7 +47,7 @@ import { annotate } from "@/lib/logging/context"; /** One mappable activity metric: a range fetcher + a body mapper + a verb. */ interface ActivityResource { fetch: (accessToken: string, start: Date, end: Date) => Promise; - map: (body: unknown) => FitbitMappedMeasurement[]; + map: (body: unknown, tz: string) => FitbitMappedMeasurement[]; verb: string; } @@ -90,8 +92,12 @@ export async function syncUserActivity( const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; + // Daily rows are anchored at local noon of their date in the user's zone, + // and the window runs through the user's own today (a UTC "now" is still + // yesterday for most of a morning in a zone ahead of UTC). + const tz = await resolveUserTimezone(userId); const start = opts.start ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); - const end = opts.end ?? new Date(); + const end = opts.end ?? localTodayAsDate(tz); const windows = chunkDateRanges(start, end, FITBIT_RANGE_DAYS); let imported = 0; @@ -112,7 +118,7 @@ export async function syncUserActivity( break; } try { - for (const m of resource.map(body)) { + for (const m of resource.map(body, tz)) { readings.push({ type: m.type, value: m.value, diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index f67f4422b..133c7fdf2 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/src/lib/fitbit/sync-core.ts @@ -24,6 +24,7 @@ import { refreshAccessToken } from "./client"; import { getUserFitbitCredentials } from "./credentials"; import { FitbitApiError, classifyFitbitError } from "./response-classifier"; import type { FitbitClassification } from "./response-classifier"; +import { userDayKey } from "@/lib/tz/format"; /** Refresh the access token this many ms before `tokenExpiresAt`. */ const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; @@ -383,6 +384,14 @@ export function incrementalStart( return new Date(now.getTime() - 30 * 24 * 60 * 60 * 1000); } +/** + * The user's local today as a Date whose UTC calendar date is that day, for + * the date-keyed Fitbit range endpoints (`fitbitDate` reads the UTC date). + */ +export function localTodayAsDate(tz: string, now: Date = new Date()): Date { + return new Date(`${userDayKey(now, tz)}T12:00:00.000Z`); +} + /** * Slice `[start, end]` into consecutive inclusive windows no wider than * `maxDays` calendar days, the per-endpoint range cap the classic Web API @@ -626,6 +635,8 @@ export async function upsertFitbitMeasurements( r: FitbitMeasurementUpsert; /** Set when a key-format migration re-keys the row in place (see below). */ reKeyTo?: string; + /** The row's instant before this write, when the write moves it. */ + movedFrom?: Date; }> = []; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; @@ -648,7 +659,16 @@ export async function upsertFitbitMeasurements( existingRow.unit === r.unit && existingRow.measuredAt.getTime() === r.measuredAt.getTime() && (existingRow.sleepStage ?? null) === (r.sleepStage ?? null); - if (!unchanged) toUpdate.push({ id: existingRow.id, r }); + if (!unchanged) { + toUpdate.push({ + id: existingRow.id, + r, + movedFrom: + existingRow.measuredAt.getTime() !== r.measuredAt.getTime() + ? existingRow.measuredAt + : undefined, + }); + } } else if (!plannedCreateKeys.has(key)) { plannedCreateKeys.add(key); toCreate.push({ @@ -813,7 +833,7 @@ export async function upsertFitbitMeasurements( // `syncVersion`. `deletedAt: null` rides along unconditionally — a no-op on // a live row, a deliberate RESURRECTION on a tombstoned one (Fitbit is the // source of truth for its own rows; see TOMBSTONES RESURRECT above). - for (const { id, r, reKeyTo } of toUpdate) { + for (const { id, r, reKeyTo, movedFrom } of toUpdate) { try { await prisma.measurement.update({ where: { id }, @@ -833,6 +853,14 @@ export async function upsertFitbitMeasurements( type: r.type as MeasurementType, measuredAt: r.measuredAt, }); + // A row that moved (a daily anchor re-cut on the user's local noon) + // leaves its old day too; that day's rollup needs the refold. + if (movedFrom) { + touched.push({ + type: r.type as MeasurementType, + measuredAt: movedFrom, + }); + } imported++; } catch (err) { getEvent()?.addWarning(`Fitbit: failed to update measurement: ${err}`); diff --git a/src/lib/fitbit/sync-metrics.ts b/src/lib/fitbit/sync-metrics.ts index 17a6365c5..066bd4a6d 100644 --- a/src/lib/fitbit/sync-metrics.ts +++ b/src/lib/fitbit/sync-metrics.ts @@ -38,6 +38,8 @@ import { mapWeight, } from "./client"; import type { FitbitMappedMeasurement } from "./client"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; +import { localTodayAsDate } from "./sync-core"; import { chunkDateRanges, getValidToken, @@ -52,7 +54,7 @@ import type { /** One mappable metric: a range fetcher + a body mapper + a verb. */ interface MetricResource { fetch: (accessToken: string, start: Date, end: Date) => Promise; - map: (body: unknown) => FitbitMappedMeasurement[]; + map: (body: unknown, tz: string) => FitbitMappedMeasurement[]; verb: string; } @@ -80,8 +82,12 @@ export async function syncUserMetrics( const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; + // Daily rows are anchored at local noon of their date in the user's zone, + // and the window runs through the user's own today (a UTC "now" is still + // yesterday for most of a morning in a zone ahead of UTC). + const tz = await resolveUserTimezone(userId); const start = opts.start ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); - const end = opts.end ?? new Date(); + const end = opts.end ?? localTodayAsDate(tz); const windows = chunkDateRanges(start, end, FITBIT_RANGE_DAYS); let imported = 0; @@ -106,7 +112,7 @@ export async function syncUserMetrics( break; } try { - for (const m of resource.map(body)) { + for (const m of resource.map(body, tz)) { readings.push({ type: m.type, value: m.value, diff --git a/src/lib/google-health/__tests__/mappers.test.ts b/src/lib/google-health/__tests__/mappers.test.ts index c32d36c88..49f3fa8ea 100644 --- a/src/lib/google-health/__tests__/mappers.test.ts +++ b/src/lib/google-health/__tests__/mappers.test.ts @@ -762,3 +762,37 @@ describe("chunkCivilRange — dailyRollUp 90-day slicing", () => { ).toEqual([]); }); }); + +describe("daily anchors in the user's zone", () => { + it.each([ + ["Pacific/Tongatapu", "2026-05-31T23:00:00.000Z"], + ["Pacific/Auckland", "2026-06-01T00:00:00.000Z"], + ["America/New_York", "2026-06-01T16:00:00.000Z"], + ["Europe/Berlin", "2026-06-01T10:00:00.000Z"], + ])( + "anchors a daily total and a daily summary at local noon in %s", + (tz, iso) => { + const [steps] = mapSteps( + { + civilStartTime: { date: { year: 2026, month: 6, day: 1 } }, + steps: { countSum: "8500" }, + }, + tz, + ); + expect(steps!.measuredAt.toISOString()).toBe(iso); + // The externalId stays on the civil date whatever the zone. + expect(steps!.fieldTag).toBe("steps:2026-06-01"); + const [spo2] = mapOxygenSaturation( + { + dailyOxygenSaturation: { + averagePercentage: 97.4, + date: { year: 2026, month: 6, day: 1 }, + }, + }, + tz, + ); + expect(spo2!.measuredAt.toISOString()).toBe(iso); + expect(spo2!.fieldTag).toBe("2026-06-01:spo2"); + }, + ); +}); diff --git a/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts b/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts index bd207e2b2..0704e346c 100644 --- a/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts +++ b/src/lib/google-health/__tests__/sync-upsert-resurrect.test.ts @@ -88,6 +88,7 @@ describe("upsertGoogleHealthMeasurements — tombstones resurrect", () => { id: "row-1", type: "ACTIVITY_STEPS", externalId: "stats:steps:2026-07-08", + measuredAt: STEPS_READING.measuredAt, }, ]); diff --git a/src/lib/google-health/mappers.ts b/src/lib/google-health/mappers.ts index a0684fbaa..a94d7abed 100644 --- a/src/lib/google-health/mappers.ts +++ b/src/lib/google-health/mappers.ts @@ -14,6 +14,7 @@ * this file performs I/O — everything is synchronous and deterministic given a * payload (+ optional user timezone). */ +import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; import { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; /** @@ -290,8 +291,19 @@ function readPath(obj: unknown, path: string): unknown { return cur; } -/** Parse a `{year,month,day}` civil-date object into a UTC-midday Date, or null. */ -function parseCivilDateObject(val: unknown): Date | null { +/** + * Read a Google civil date — a `{year,month,day}` object, or a civil string + * (`YYYY-MM-DD`, optionally with a time suffix like `2026-06-02T00:00:00`) — + * as a `YYYY-MM-DD` day key, or null. The key names the row's externalId; the + * row's instant is the key's local noon in the user's zone + * (`canonicalDailyTimestamp`), which reads the same calendar day there. Noon + * UTC, the previous anchor, is already the next day from UTC+13 on. + */ +function parseCivilKey(val: unknown): string | null { + if (typeof val === "string") { + const m = /^(\d{4})-(\d{2})-(\d{2})/.exec(val.trim()); + return m ? `${m[1]}-${m[2]}-${m[3]}` : null; + } if (!val || typeof val !== "object") return null; const o = val as Record; if ( @@ -299,30 +311,12 @@ function parseCivilDateObject(val: unknown): Date | null { typeof o.month === "number" && typeof o.day === "number" ) { - // Google civil dates are 1-based months; anchor at UTC midday so a timezone - // shift can't roll the civil day across a boundary. - return new Date(Date.UTC(o.year, o.month - 1, o.day, 12)); + const pad = (n: number, w: number) => String(n).padStart(w, "0"); + return `${pad(o.year, 4)}-${pad(o.month, 2)}-${pad(o.day, 2)}`; } return null; } -/** - * Parse a Google civil START anchor — the calendar day a cumulative daily total - * belongs to — into a UTC-midday Date, or null. Accepts a `{year,month,day}` - * object OR a civil string (`YYYY-MM-DD`, optionally carrying a time suffix like - * `2026-06-02T00:00:00`); only the Y-M-D is kept, anchored at UTC midday so a - * timezone shift can't roll the civil day. Mirrors the Fitbit `parseCivilDate` - * UTC-midday convention. - */ -function parseCivilStart(val: unknown): Date | null { - if (typeof val === "string") { - const m = /^(\d{4})-(\d{2})-(\d{2})/.exec(val.trim()); - if (!m) return null; - return new Date(Date.UTC(Number(m[1]), Number(m[2]) - 1, Number(m[3]), 12)); - } - return parseCivilDateObject(val); -} - /** * Matches an offset-less local ISO wall-clock string (`2026-06-02T03:02:30` / * `...T03:02:30.000`). No trailing `Z`, no `±hh:mm` — those denote an absolute @@ -369,7 +363,7 @@ function parseLocalInstant(iso: string, tz?: string): Date | null { * - `sample` → `{key}.sampleTime.physicalTime` (a spot instant; offset-less * strings resolve against `tz`). * - `date` → `{key}.date` (a `{year,month,day}` object, or a civil string) — - * anchored at UTC-midday so a tz shift can't roll the civil day. + * anchored at local noon of that day in the user's zone. * Falls back to `fallback` only when nothing parses, so a row is never dropped * for a missing anchor. (Sleep / exercise sessions and the rollup types anchor * through their own helpers, never here.) @@ -387,9 +381,8 @@ function resolveMeasuredAt( if (d) return d; } } else if (dataType.timeField === "date") { - const dateVal = readPath(point, `${dataType.key}.date`); - const civil = parseCivilStart(dateVal); - if (civil) return civil; + const day = parseCivilKey(readPath(point, `${dataType.key}.date`)); + if (day) return canonicalDailyTimestamp(day, tz); } return fallback; } @@ -404,14 +397,14 @@ function externalAnchor( dataType: GoogleHealthDataType, tz?: string, ): string { - const at = resolveMeasuredAt(point, dataType, new Date(0), tz); // Daily summaries share the civil-day externalId grain so a re-fetched day // overwrites in place. (Sleep/exercise sessions and the rollup daily totals // mint their own anchors and never reach this helper.) if (dataType.timeField === "date") { - return at.toISOString().slice(0, 10); + const day = parseCivilKey(readPath(point, `${dataType.key}.date`)); + if (day) return day; } - return at.toISOString(); + return resolveMeasuredAt(point, dataType, new Date(0), tz).toISOString(); } /** @@ -429,6 +422,7 @@ function mapSimple( valuePaths: string[]; factor?: number; }, + tz: string | undefined, ): GoogleHealthMappedMeasurement[] { let value = firstNumber(point, spec.valuePaths); if (value === null) return []; @@ -438,7 +432,7 @@ function mapSimple( type: spec.type, value: round2(value), unit: spec.unit, - measuredAt: resolveMeasuredAt(point, dataType, new Date()), + measuredAt: resolveMeasuredAt(point, dataType, new Date(), tz), fieldTag: `${externalAnchor(point, dataType)}:${spec.fieldTag}`, }, ]; @@ -446,44 +440,63 @@ function mapSimple( export function mapWeight( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.weight; // Documented field is `weightGrams` (grams) → kg. - return mapSimple(point, dt, { - type: "WEIGHT", - unit: "kg", - fieldTag: "weight", - valuePaths: [`${dt.key}.weightGrams`], - factor: 0.001, - }); + return mapSimple( + point, + dt, + { + type: "WEIGHT", + unit: "kg", + fieldTag: "weight", + valuePaths: [`${dt.key}.weightGrams`], + factor: 0.001, + }, + tz, + ); } export function mapBodyFat( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.bodyFat; - return mapSimple(point, dt, { - type: "BODY_FAT", - unit: "%", - fieldTag: "body_fat", - valuePaths: [`${dt.key}.percentage`], - }); + return mapSimple( + point, + dt, + { + type: "BODY_FAT", + unit: "%", + fieldTag: "body_fat", + valuePaths: [`${dt.key}.percentage`], + }, + tz, + ); } export function mapOxygenSaturation( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.oxygenSaturation; - return mapSimple(point, dt, { - type: "OXYGEN_SATURATION", - unit: "%", - fieldTag: "spo2", - valuePaths: [`${dt.key}.averagePercentage`], - }); + return mapSimple( + point, + dt, + { + type: "OXYGEN_SATURATION", + unit: "%", + fieldTag: "spo2", + valuePaths: [`${dt.key}.averagePercentage`], + }, + tz, + ); } export function mapHeartRateVariability( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.heartRateVariability; // The daily field is an unlabelled "average HRV ms". Per the design decision @@ -491,72 +504,102 @@ export function mapHeartRateVariability( // (Apple-comparable), NOT WHOOP's `HRV_RMSSD` (reserved for the WHOOP-native // estimator). The per-sample type carries explicit RMSSD + SDNN fields — // re-confirm the estimator against live data and revisit if warranted. - return mapSimple(point, dt, { - type: "HEART_RATE_VARIABILITY", - unit: "ms", - fieldTag: "hrv", - valuePaths: [`${dt.key}.averageHeartRateVariabilityMilliseconds`], - }); + return mapSimple( + point, + dt, + { + type: "HEART_RATE_VARIABILITY", + unit: "ms", + fieldTag: "hrv", + valuePaths: [`${dt.key}.averageHeartRateVariabilityMilliseconds`], + }, + tz, + ); } export function mapRestingHeartRate( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.restingHeartRate; // `beatsPerMinute` is an int64 JSON string — coerced by the extractor. - return mapSimple(point, dt, { - type: "RESTING_HEART_RATE", - unit: "bpm", - fieldTag: "rhr", - valuePaths: [`${dt.key}.beatsPerMinute`], - }); + return mapSimple( + point, + dt, + { + type: "RESTING_HEART_RATE", + unit: "bpm", + fieldTag: "rhr", + valuePaths: [`${dt.key}.beatsPerMinute`], + }, + tz, + ); } export function mapRespiratoryRate( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.respiratoryRate; // Documented schema: `{ date, breathsPerMinute }` — `breathsPerMinute` is a // plain number ("The average number of breaths taken per minute"). The // earlier `dailyRespiratoryRateBpm` leaf does not exist and never parsed. - return mapSimple(point, dt, { - type: "RESPIRATORY_RATE", - unit: "breaths/min", - fieldTag: "resp_rate", - valuePaths: [`${dt.key}.breathsPerMinute`], - }); + return mapSimple( + point, + dt, + { + type: "RESPIRATORY_RATE", + unit: "breaths/min", + fieldTag: "resp_rate", + valuePaths: [`${dt.key}.breathsPerMinute`], + }, + tz, + ); } export function mapBloodGlucose( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.bloodGlucose; // Documented field `bloodGlucoseMilligramsPerDeciliter` (number) — already in // HealthLog's canonical mg/dL storage unit, no conversion. - return mapSimple(point, dt, { - type: "BLOOD_GLUCOSE", - unit: "mg/dL", - fieldTag: "glucose", - valuePaths: [`${dt.key}.bloodGlucoseMilligramsPerDeciliter`], - }); + return mapSimple( + point, + dt, + { + type: "BLOOD_GLUCOSE", + unit: "mg/dL", + fieldTag: "glucose", + valuePaths: [`${dt.key}.bloodGlucoseMilligramsPerDeciliter`], + }, + tz, + ); } export function mapCoreBodyTemperature( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.coreBodyTemperature; // Documented field `temperatureCelsius` (number) → the core BODY_TEMPERATURE // slot (distinct from SKIN_TEMPERATURE / WRIST_TEMPERATURE surface readings). - return mapSimple(point, dt, { - type: "BODY_TEMPERATURE", - unit: "celsius", - fieldTag: "core_temp", - valuePaths: [`${dt.key}.temperatureCelsius`], - }); + return mapSimple( + point, + dt, + { + type: "BODY_TEMPERATURE", + unit: "celsius", + fieldTag: "core_temp", + valuePaths: [`${dt.key}.temperatureCelsius`], + }, + tz, + ); } export function mapWristTemperature( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.sleepTemperature; // `nightlyTemperatureCelsius` is the ABSOLUTE nightly skin temperature ("the @@ -565,25 +608,36 @@ export function mapWristTemperature( // absolute-reading convention. The sibling `baselineTemperatureCelsius` / // `relativeNightlyStddev30dCelsius` derivations are not stored — the user's // own series carries the baseline. - return mapSimple(point, dt, { - type: "WRIST_TEMPERATURE", - unit: "celsius", - fieldTag: "wrist_temp", - valuePaths: [`${dt.key}.nightlyTemperatureCelsius`], - }); + return mapSimple( + point, + dt, + { + type: "WRIST_TEMPERATURE", + unit: "celsius", + fieldTag: "wrist_temp", + valuePaths: [`${dt.key}.nightlyTemperatureCelsius`], + }, + tz, + ); } export function mapHeartRate( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.heartRate; // `beatsPerMinute` is an int64 JSON string. - return mapSimple(point, dt, { - type: "PULSE", - unit: "bpm", - fieldTag: "hr", - valuePaths: [`${dt.key}.beatsPerMinute`], - }); + return mapSimple( + point, + dt, + { + type: "PULSE", + unit: "bpm", + fieldTag: "hr", + valuePaths: [`${dt.key}.beatsPerMinute`], + }, + tz, + ); } /** One extracted `height` sample: centimetres + the sample instant (if any). */ @@ -635,8 +689,8 @@ function firstNonNegativeNumber( /** * Map one `dailyRollUp` aggregate window (`windowSizeDays: 1`) into a single * daily-total Measurement reading. The day key comes from the window's - * `civilStartTime.date` (a `{year,month,day}` object), anchored at UTC-midday - * per the shared civil-day convention; a window with no parseable civil day is + * `civilStartTime.date` (a `{year,month,day}` object), anchored at local noon + * of that day in the user's zone; a window with no parseable civil day is * dropped (it cannot be keyed). The externalId is the `stats:`-prefixed * daily-total shape so a re-fetched day overwrites in place — the same * overwrite contract the Apple-Health `stats::` daily totals @@ -651,21 +705,21 @@ function mapDailyRollup( valuePaths: string[]; factor?: number; }, + tz: string | undefined, ): GoogleHealthMappedMeasurement[] { - const day = - parseCivilDateObject(readPath(point, "civilStartTime.date")) ?? - parseCivilStart(readPath(point, "civilStartTime")); - if (!day) return []; + const dayKey = + parseCivilKey(readPath(point, "civilStartTime.date")) ?? + parseCivilKey(readPath(point, "civilStartTime")); + if (!dayKey) return []; let value = firstNonNegativeNumber(point, spec.valuePaths); if (value === null) return []; if (spec.factor) value = value * spec.factor; - const dayKey = day.toISOString().slice(0, 10); return [ { type: spec.type, value: round2(value), unit: spec.unit, - measuredAt: day, + measuredAt: canonicalDailyTimestamp(dayKey, tz), // `stats::` — the sync layer reads `cumulativeDaily` // to assemble the externalId, matching the Apple-Health daily-total shape. fieldTag: `${spec.fieldTag}:${dayKey}`, @@ -676,55 +730,75 @@ function mapDailyRollup( export function mapSteps( point: GoogleHealthRollupPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.steps; // `countSum` is an int64 JSON string. - return mapDailyRollup(point, { - type: "ACTIVITY_STEPS", - unit: "steps", - fieldTag: "steps", - valuePaths: [`${dt.key}.countSum`], - }); + return mapDailyRollup( + point, + { + type: "ACTIVITY_STEPS", + unit: "steps", + fieldTag: "steps", + valuePaths: [`${dt.key}.countSum`], + }, + tz, + ); } export function mapDistance( point: GoogleHealthRollupPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.distance; // `millimetersSum` is an int64 JSON string → metres. - return mapDailyRollup(point, { - type: "WALKING_RUNNING_DISTANCE", - unit: "m", - fieldTag: "distance", - valuePaths: [`${dt.key}.millimetersSum`], - factor: 0.001, - }); + return mapDailyRollup( + point, + { + type: "WALKING_RUNNING_DISTANCE", + unit: "m", + fieldTag: "distance", + valuePaths: [`${dt.key}.millimetersSum`], + factor: 0.001, + }, + tz, + ); } export function mapActiveEnergy( point: GoogleHealthRollupPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.activeEnergy; // ACTIVE energy only — NOT total-calories (which folds in BMR). - return mapDailyRollup(point, { - type: "ACTIVE_ENERGY_BURNED", - unit: "kcal", - fieldTag: "active_energy", - valuePaths: [`${dt.key}.kcalSum`], - }); + return mapDailyRollup( + point, + { + type: "ACTIVE_ENERGY_BURNED", + unit: "kcal", + fieldTag: "active_energy", + valuePaths: [`${dt.key}.kcalSum`], + }, + tz, + ); } export function mapFloors( point: GoogleHealthRollupPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.floors; // `countSum` is an int64 JSON string. - return mapDailyRollup(point, { - type: "FLIGHTS_CLIMBED", - unit: "flights", - fieldTag: "floors", - valuePaths: [`${dt.key}.countSum`], - }); + return mapDailyRollup( + point, + { + type: "FLIGHTS_CLIMBED", + unit: "flights", + fieldTag: "floors", + valuePaths: [`${dt.key}.countSum`], + }, + tz, + ); } /** @@ -735,11 +809,12 @@ export function mapFloors( */ export function mapVo2Max( point: GoogleHealthDataPoint, + tz?: string, ): GoogleHealthMappedMeasurement[] { const dt = GOOGLE_HEALTH_DATA_TYPES.vo2Max; const value = firstNumber(point, [`${dt.key}.vo2Max`]); if (value === null) return []; - const measuredAt = resolveMeasuredAt(point, dt, new Date()); + const measuredAt = resolveMeasuredAt(point, dt, new Date(), tz); return [ { type: "VO2_MAX", diff --git a/src/lib/google-health/sync-activity.ts b/src/lib/google-health/sync-activity.ts index 8f40c9ca8..80cd1be80 100644 --- a/src/lib/google-health/sync-activity.ts +++ b/src/lib/google-health/sync-activity.ts @@ -49,7 +49,10 @@ import { resolveUserTimezone } from "@/lib/tz/resolver"; /** One mappable activity metric: its data-type encoding + the mapper + a verb. */ interface ActivityResource { dataType: GoogleHealthDataType; - map: (point: Record) => GoogleHealthMappedMeasurement[]; + map: ( + point: Record, + tz: string, + ) => GoogleHealthMappedMeasurement[]; verb: string; } @@ -96,8 +99,8 @@ export async function syncUserActivity( // The dailyRollUp request range is civil and user-local; resolve the user's // stored zone so the range bounds land on the correct civil days rather than - // the process zone's. (The response day-key comes from each window's own - // `civilStartTime.date`, tz-independent.) + // the process zone's. The response day-key comes from each window's own + // `civilStartTime.date`; the row is anchored at local noon of that day. const tz = await resolveUserTimezone(userId); // Cycle-wide watermark snapshotted once by `syncUserGoogleHealth`; undefined @@ -136,7 +139,7 @@ export async function syncUserActivity( const readings: GoogleHealthMeasurementUpsert[] = []; try { for (const point of points) { - for (const m of resource.map(point)) { + for (const m of resource.map(point, tz)) { readings.push({ type: m.type, value: m.value, @@ -180,7 +183,7 @@ export async function syncUserActivity( ); const readings: GoogleHealthMeasurementUpsert[] = []; for (const point of points) { - for (const m of mapVo2Max(point)) { + for (const m of mapVo2Max(point, tz)) { readings.push({ type: m.type, value: m.value, diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index d0545ee0d..bef34d971 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -585,6 +585,8 @@ export async function upsertGoogleHealthMeasurements( r: GoogleHealthMeasurementUpsert; /** Set when a key-format migration re-keys the row in place (see below). */ reKeyTo?: string; + /** The row's instant before this write, when the write moves it. */ + movedFrom?: Date; }> = []; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; @@ -607,7 +609,16 @@ export async function upsertGoogleHealthMeasurements( existingRow.unit === r.unit && existingRow.measuredAt.getTime() === r.measuredAt.getTime() && (existingRow.sleepStage ?? null) === (r.sleepStage ?? null); - if (!unchanged) toUpdate.push({ id: existingRow.id, r }); + if (!unchanged) { + toUpdate.push({ + id: existingRow.id, + r, + movedFrom: + existingRow.measuredAt.getTime() !== r.measuredAt.getTime() + ? existingRow.measuredAt + : undefined, + }); + } } else if (!plannedCreateKeys.has(key)) { plannedCreateKeys.add(key); toCreate.push({ @@ -776,7 +787,7 @@ export async function upsertGoogleHealthMeasurements( // `syncVersion`. `deletedAt: null` rides along unconditionally — a no-op on // a live row, a deliberate RESURRECTION on a tombstoned one (Google is the // source of truth for its own rows; see TOMBSTONES RESURRECT above). - for (const { id, r, reKeyTo } of toUpdate) { + for (const { id, r, reKeyTo, movedFrom } of toUpdate) { try { await prisma.measurement.update({ where: { id }, @@ -796,6 +807,14 @@ export async function upsertGoogleHealthMeasurements( type: r.type as MeasurementType, measuredAt: r.measuredAt, }); + // A row that moved (a daily anchor re-cut on the user's local noon) + // leaves its old day too; that day's rollup needs the refold. + if (movedFrom) { + touched.push({ + type: r.type as MeasurementType, + measuredAt: movedFrom, + }); + } imported++; } catch (err) { getEvent()?.addWarning( diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index 85ad5951d..6c204c7ab 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -51,11 +51,15 @@ import { } from "./sync-core"; import { getEvent } from "@/lib/logging/context"; import { prisma } from "@/lib/db"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; /** One mappable metric: its data-type encoding + the per-point mapper + a verb. */ interface MetricResource { dataType: GoogleHealthDataType; - map: (point: Record) => GoogleHealthMappedMeasurement[]; + map: ( + point: Record, + tz: string, + ) => GoogleHealthMappedMeasurement[]; verb: string; } @@ -127,6 +131,9 @@ export async function syncUserMetrics( // `syncUserGoogleHealth` — never re-read here, so a sibling resource's stamp // can't shrink this one's window. Undefined on a full/backfill run. const start = opts.start; + // Daily summaries are anchored at local noon of their date, and offset-less + // sample times are read on the user's wall clock. + const tz = await resolveUserTimezone(userId); let imported = 0; @@ -155,7 +162,7 @@ export async function syncUserMetrics( const readings: GoogleHealthMeasurementUpsert[] = []; try { for (const point of points) { - for (const m of resource.map(point)) { + for (const m of resource.map(point, tz)) { readings.push({ type: m.type, value: m.value, diff --git a/src/lib/measurements/consolidation-tz.ts b/src/lib/measurements/consolidation-tz.ts index de34efc92..133cf4f16 100644 --- a/src/lib/measurements/consolidation-tz.ts +++ b/src/lib/measurements/consolidation-tz.ts @@ -70,9 +70,9 @@ export function dayKeyForUserTz(date: Date, tz: string): string { /** * Compute the canonical timestamp for a calendar-day key. With a `tz`, * returns the JS-Date instant at the user's local 12:00 noon; without - * one, returns 12:00 UTC of the day. Either anchor sits a full 12 h - * inside its calendar day, so it round-trips back to the same day - * through `userDayKey()` for every zone within ±12 h — which is the + * one, returns 12:00 UTC of the day. The local-noon anchor sits a full + * 12 h inside its calendar day, so it round-trips back to the same day + * through `userDayKey()` in that zone — which is the * whole point of anchoring date-only daily records at noon rather than * midnight (a UTC-midnight anchor double-shifts the day for west-of-UTC * users on read). Matches the Withings activity sync convention (one @@ -80,10 +80,11 @@ export function dayKeyForUserTz(date: Date, tz: string): string { * between same-day spot samples). The string returned by * `toISOString()` is UTC. * - * The optional-`tz` shape is what the wearable daily mappers (Oura, - * Polar, Fitbit) anchor on — they do not have the user's timezone in - * scope at map time, so they take the noon-UTC fallback, which is - * correct for every zone within ±12 h. + * Every daily row a sync writes (Withings, Fitbit, Google Health, Polar, + * Oura) passes the user's zone. The noon-UTC fallback is NOT day-stable + * everywhere: from UTC+12 on (New Zealand all year, Fiji, Tonga, Samoa) noon + * UTC is already the next local day, which is how those users saw every + * synced daily total a day late up to v1.39.2. */ export function canonicalDailyTimestamp(dateKey: string, tz?: string): Date { // No timezone in scope: 12:00 UTC is a safe day-stable anchor for diff --git a/src/lib/oura/__tests__/client.test.ts b/src/lib/oura/__tests__/client.test.ts index ae2897107..1fead0330 100644 --- a/src/lib/oura/__tests__/client.test.ts +++ b/src/lib/oura/__tests__/client.test.ts @@ -642,3 +642,23 @@ describe("derivePeriodDaysFromCyclePhases", () => { expect(days).toEqual(["2026-06-10"]); }); }); + +describe("date-only records in the user's zone", () => { + it.each([ + ["Pacific/Tongatapu", "2026-05-09T23:00:00.000Z"], + ["Pacific/Auckland", "2026-05-10T00:00:00.000Z"], + ["America/New_York", "2026-05-10T16:00:00.000Z"], + ["Europe/Berlin", "2026-05-10T10:00:00.000Z"], + ])("anchors a date-only record at local noon in %s", (tz, iso) => { + const [spo2] = mapDailySpo2( + { day: "2026-05-10", spo2_percentage: { average: 97 } } as never, + tz, + ); + expect(spo2!.measuredAt.toISOString()).toBe(iso); + const [resilience] = mapResilience( + { day: "2026-05-10", level: "solid" } as never, + tz, + ); + expect(resilience!.measuredAt.toISOString()).toBe(iso); + }); +}); diff --git a/src/lib/oura/client.ts b/src/lib/oura/client.ts index 6758b2901..ede36d3e6 100644 --- a/src/lib/oura/client.ts +++ b/src/lib/oura/client.ts @@ -512,17 +512,20 @@ export interface MappedMeasurement { externalId?: string; } -function dayAnchor(day: string, bedtimeEnd?: string): Date | null { +function dayAnchor( + day: string, + tz: string | undefined, + bedtimeEnd?: string, +): Date | null { // Prefer the precise wake instant when present; else anchor the date-only - // day at noon (canonicalDailyTimestamp), NOT UTC midnight — a midnight - // anchor double-shifts the calendar day for west-of-UTC users when the - // recovery resolver re-buckets via userDayKey(). Noon sits a full 12 h - // inside the day, so it round-trips to the same day for every zone ±12 h. + // day at local noon in the user's zone (canonicalDailyTimestamp), which + // reads the same calendar day there. Noon UTC, the previous anchor, is + // already the next day from UTC+13 on. if (bedtimeEnd) { const d = new Date(bedtimeEnd); return Number.isNaN(d.getTime()) ? null : d; } - const d = canonicalDailyTimestamp(day); + const d = canonicalDailyTimestamp(day, tz); return Number.isNaN(d.getTime()) ? null : d; } @@ -539,8 +542,11 @@ function dayAnchor(day: string, bedtimeEnd?: string): Date | null { * `temperature_trend_deviation` is the long-term trend twin; we keep only the * nightly deviation as the actionable signal. */ -export function mapReadiness(r: OuraReadiness): MappedMeasurement[] { - const measuredAt = dayAnchor(r.day, r.timestamp); +export function mapReadiness( + r: OuraReadiness, + tz?: string, +): MappedMeasurement[] { + const measuredAt = dayAnchor(r.day, tz, r.timestamp); if (!measuredAt) return []; const out: MappedMeasurement[] = []; if (typeof r.score === "number") { @@ -648,8 +654,8 @@ function mapSleepTimeline(s: OuraSleep): MappedMeasurement[] | null { * sleep END. Either way we add efficiency, HRV, RHR (`lowest_heart_rate`), and * respiratory rate. */ -export function mapSleep(s: OuraSleep): MappedMeasurement[] { - const measuredAt = dayAnchor(s.day, s.bedtime_end); +export function mapSleep(s: OuraSleep, tz?: string): MappedMeasurement[] { + const measuredAt = dayAnchor(s.day, tz, s.bedtime_end); if (!measuredAt) return []; const out: MappedMeasurement[] = []; @@ -729,9 +735,12 @@ export function mapSleep(s: OuraSleep): MappedMeasurement[] { } /** Map one Oura daily-sleep record: the headline Sleep Score → `SLEEP_SCORE`. */ -export function mapDailySleep(d: OuraDailySleep): MappedMeasurement[] { +export function mapDailySleep( + d: OuraDailySleep, + tz?: string, +): MappedMeasurement[] { if (typeof d.score !== "number") return []; - const measuredAt = dayAnchor(d.day, d.timestamp); + const measuredAt = dayAnchor(d.day, tz, d.timestamp); if (!measuredAt) return []; return [ { @@ -745,12 +754,15 @@ export function mapDailySleep(d: OuraDailySleep): MappedMeasurement[] { } /** Map one Oura daily-spo2 record: average overnight SpO2 → `OXYGEN_SATURATION`. */ -export function mapDailySpo2(s: OuraDailySpo2): MappedMeasurement[] { +export function mapDailySpo2( + s: OuraDailySpo2, + tz?: string, +): MappedMeasurement[] { const avg = s.spo2_percentage?.average; if (typeof avg !== "number" || avg <= 0) return []; - // Anchor SpO2 at the day's UTC midnight (no per-record instant in the + // Anchor SpO2 at the day's local noon (no per-record instant in the // daily_spo2 collection); the recovery / rollup readers key off the local day. - const measuredAt = dayAnchor(s.day); + const measuredAt = dayAnchor(s.day, tz); if (!measuredAt) return []; return [ { @@ -765,8 +777,11 @@ export function mapDailySpo2(s: OuraDailySpo2): MappedMeasurement[] { /** Map one Oura daily-activity record: steps, active energy, and the * equivalent walking distance (B5 — the OURA distance ladder slot was dead). */ -export function mapDailyActivity(a: OuraDailyActivity): MappedMeasurement[] { - const measuredAt = dayAnchor(a.day, a.timestamp); +export function mapDailyActivity( + a: OuraDailyActivity, + tz?: string, +): MappedMeasurement[] { + const measuredAt = dayAnchor(a.day, tz, a.timestamp); if (!measuredAt) return []; const out: MappedMeasurement[] = []; if (typeof a.steps === "number" && a.steps >= 0) { @@ -809,9 +824,9 @@ export function mapDailyActivity(a: OuraDailyActivity): MappedMeasurement[] { * `VO2_MAX` (mL/(kg·min), the canonical DB unit — no conversion). Skips a * record with no positive value. */ -export function mapVo2Max(v: OuraVo2Max): MappedMeasurement[] { +export function mapVo2Max(v: OuraVo2Max, tz?: string): MappedMeasurement[] { if (typeof v.vo2_max !== "number" || v.vo2_max <= 0) return []; - const measuredAt = dayAnchor(v.day, v.timestamp); + const measuredAt = dayAnchor(v.day, tz, v.timestamp); if (!measuredAt) return []; return [ { @@ -832,9 +847,10 @@ export function mapVo2Max(v: OuraVo2Max): MappedMeasurement[] { */ export function mapCardiovascularAge( c: OuraCardiovascularAge, + tz?: string, ): MappedMeasurement[] { if (typeof c.vascular_age !== "number" || c.vascular_age <= 0) return []; - const measuredAt = dayAnchor(c.day); + const measuredAt = dayAnchor(c.day, tz); if (!measuredAt) return []; return [ { @@ -870,14 +886,17 @@ export const RESILIENCE_UNIT = "level" as const; * (limited=1 … exceptional=5) into the numeric `value`. An unknown / missing * level string mints NO row (skipped, never coerced to 0) so a future Oura band * we do not recognise never lands as a misleading reading. Anchored at the - * day's UTC midnight (the collection carries no per-record instant). + * day's local noon (the collection carries no per-record instant). */ -export function mapResilience(r: OuraResilience): MappedMeasurement[] { +export function mapResilience( + r: OuraResilience, + tz?: string, +): MappedMeasurement[] { const level = typeof r.level === "string" ? r.level.toLowerCase() : null; if (!level) return []; const ordinal = RESILIENCE_LEVELS[level]; if (typeof ordinal !== "number") return []; - const measuredAt = dayAnchor(r.day); + const measuredAt = dayAnchor(r.day, tz); if (!measuredAt) return []; return [ { diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index 8fb9bae01..b545206b0 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -90,6 +90,8 @@ import { } from "./credentials"; import { OuraApiError, classifyOuraError } from "./response-classifier"; import { syncUserOuraCyclePhases } from "./cycle-sync"; +import { shiftDateKey, userDayKey } from "@/lib/tz/format"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; /** Floor of the lookback window (days) for an incremental sync. Oura finalises * a night's scores hours after wake; 7 days re-fetches a handful of records @@ -194,6 +196,7 @@ export interface OuraMeasurementUpsert { function toUpsert( mapped: MappedMeasurement[], resourcePrefix: string, + dateOnlyDay?: string, ): OuraMeasurementUpsert[] { return mapped.map((m) => ({ type: m.type, @@ -202,9 +205,13 @@ function toUpsert( measuredAt: m.measuredAt, // A mapper that needs a record-scoped key (sleep rows — per-segment timeline // + nightly scalars) carries its own externalId; everything else falls back - // to the day-keyed `::` shape. + // to the day-keyed `::` shape. A record with no + // instant of its own (`dateOnlyDay`) keys on its date: its row sits at + // local noon in the user's zone, which east of UTC+12 is the previous UTC + // date. A timed record keys on its instant's UTC date, as it always has. externalId: - m.externalId ?? `${resourcePrefix}:${ymd(m.measuredAt)}:${m.fieldTag}`, + m.externalId ?? + `${resourcePrefix}:${dateOnlyDay ?? ymd(m.measuredAt)}:${m.fieldTag}`, sleepStage: m.sleepStage ?? null, })); } @@ -260,9 +267,15 @@ async function fetchAll( refreshTokenCiphertext: string, lookbackDays: number, ): Promise { - const now = new Date(); - const start = new Date(now.getTime() - lookbackDays * 24 * 60 * 60 * 1000); - const query = { startDate: ymd(start), endDate: ymd(now) }; + // Oura keys its daily collections by the user's calendar date, so the + // window is cut on the user's own days, and date-only records are anchored + // at local noon of their date in this zone. + const tz = await resolveUserTimezone(userId); + const today = userDayKey(new Date(), tz); + const query = { + startDate: shiftDateKey(today, -lookbackDays), + endDate: today, + }; // Filled by the sleep collection's closure as it maps records; carried on // the result so the caller can run the record-scoped sweep before the @@ -275,7 +288,11 @@ async function fetchAll( name: "readiness", collect: async (t) => (await fetchReadiness(t, query)).flatMap((r) => - toUpsert(mapReadiness(r), "readiness"), + toUpsert( + mapReadiness(r, tz), + "readiness", + r.timestamp ? undefined : r.day, + ), ), }, { @@ -284,7 +301,11 @@ async function fetchAll( const records = await fetchSleep(t, query); const ups: OuraMeasurementUpsert[] = []; for (const s of records) { - const rows = toUpsert(mapSleep(s), "sleep"); + const rows = toUpsert( + mapSleep(s, tz), + "sleep", + s.bedtime_end ? undefined : s.day, + ); ups.push(...rows); // Record-scoped sweep entry: every SLEEP_DURATION row of this // record keys under `sleep::` (hypnogram runs AND the @@ -306,42 +327,50 @@ async function fetchAll( name: "activity", collect: async (t) => (await fetchDailyActivity(t, query)).flatMap((a) => - toUpsert(mapDailyActivity(a), "activity"), + toUpsert( + mapDailyActivity(a, tz), + "activity", + a.timestamp ? undefined : a.day, + ), ), }, { name: "daily_sleep", collect: async (t) => (await fetchDailySleep(t, query)).flatMap((d) => - toUpsert(mapDailySleep(d), "daily_sleep"), + toUpsert( + mapDailySleep(d, tz), + "daily_sleep", + d.timestamp ? undefined : d.day, + ), ), }, { name: "spo2", collect: async (t) => (await fetchDailySpo2(t, query)).flatMap((s) => - toUpsert(mapDailySpo2(s), "spo2"), + toUpsert(mapDailySpo2(s, tz), "spo2", s.day), ), }, { name: "vo2max", collect: async (t) => (await fetchVo2Max(t, query)).flatMap((v) => - toUpsert(mapVo2Max(v), "vo2max"), + toUpsert(mapVo2Max(v, tz), "vo2max", v.timestamp ? undefined : v.day), ), }, { name: "cardio_age", collect: async (t) => (await fetchCardiovascularAge(t, query)).flatMap((c) => - toUpsert(mapCardiovascularAge(c), "cardio_age"), + toUpsert(mapCardiovascularAge(c, tz), "cardio_age", c.day), ), }, { name: "resilience", collect: async (t) => (await fetchResilience(t, query)).flatMap((r) => - toUpsert(mapResilience(r), "resilience"), + toUpsert(mapResilience(r, tz), "resilience", r.day), ), }, ]; diff --git a/src/lib/polar/__tests__/client.test.ts b/src/lib/polar/__tests__/client.test.ts index a92276f8f..97f3c0702 100644 --- a/src/lib/polar/__tests__/client.test.ts +++ b/src/lib/polar/__tests__/client.test.ts @@ -609,3 +609,28 @@ describe("POLAR_OAUTH_SCOPE", () => { expect(POLAR_OAUTH_SCOPE).toBe("accesslink.read_all"); }); }); + +describe("date-keyed rows in the user's zone", () => { + it.each([ + ["Pacific/Tongatapu", "2026-05-09T23:00:00.000Z"], + ["Pacific/Auckland", "2026-05-10T00:00:00.000Z"], + ["America/New_York", "2026-05-10T16:00:00.000Z"], + ["Europe/Berlin", "2026-05-10T10:00:00.000Z"], + ])( + "anchors a day's activity at local noon in %s and names the day", + (tz, iso) => { + const rows = mapActivity( + { start_time: "2026-05-10T00:00:00", steps: 9000 } as never, + tz, + ); + expect(rows[0]!.measuredAt.toISOString()).toBe(iso); + expect(rows[0]!.day).toBe("2026-05-10"); + const [load] = mapCardioLoad( + { date: "2026-05-10", cardio_load: 12 } as never, + tz, + ); + expect(load!.measuredAt.toISOString()).toBe(iso); + expect(load!.day).toBe("2026-05-10"); + }, + ); +}); diff --git a/src/lib/polar/client.ts b/src/lib/polar/client.ts index 4a6092a06..9739329ce 100644 --- a/src/lib/polar/client.ts +++ b/src/lib/polar/client.ts @@ -422,6 +422,13 @@ export interface MappedMeasurement { * sync layer uses it verbatim. */ externalId?: string; + /** + * The calendar date a date-keyed row belongs to (`YYYY-MM-DD`). The sync + * layer keys the externalId on it rather than on `measuredAt`, which sits at + * the date's local noon in the user's zone and so can fall on a different + * UTC date. + */ + day?: string; /** * `true` on per-segment sleep rows whose ORDER is synthesised. When Polar * gives only per-stage duration totals (no per-stage onset timestamps), the @@ -444,13 +451,14 @@ function rechargeStatusToScore(status: number): number { * Map one Polar Nightly Recharge record. The recovery band → `RECOVERY_SCORE` * (source POLAR, distinct from the COMPUTED proxy); HRV → `HRV_RMSSD`; the * nightly average HR → `RESTING_HEART_RATE`; breathing rate → `RESPIRATORY_RATE`. - * `measuredAt` is the record's date anchored at noon (the wake - * morning the night belongs to). + * `measuredAt` is the record's date anchored at local noon in the user's + * zone (the wake morning the night belongs to). */ export function mapNightlyRecharge( r: PolarNightlyRecharge, + tz?: string, ): MappedMeasurement[] { - const measuredAt = canonicalDailyTimestamp(r.date); + const measuredAt = canonicalDailyTimestamp(r.date, tz); if (Number.isNaN(measuredAt.getTime())) return []; const out: MappedMeasurement[] = []; @@ -507,7 +515,7 @@ export function mapNightlyRecharge( fieldTag: "ans_charge", }); } - return out; + return out.map((m) => ({ ...m, day: r.date })); } /** @@ -534,7 +542,7 @@ export function mapNightlyRecharge( * fall back to a midnight-UTC anchor and emit untimed stage rows — degraded but * never wrong-day for the common UTC-positive case, matching the pre-fix shape. */ -export function mapSleep(s: PolarSleep): MappedMeasurement[] { +export function mapSleep(s: PolarSleep, tz?: string): MappedMeasurement[] { const out: MappedMeasurement[] = []; const startMs = s.sleep_start_time ? Date.parse(s.sleep_start_time) : NaN; @@ -599,10 +607,10 @@ export function mapSleep(s: PolarSleep): MappedMeasurement[] { return out; } - // Fallback: no usable window. Anchor at noon of the wake date and emit - // untimed stage rows (the pre-v1.17.1 shape) so a record missing the window - // fields still contributes a night total. - const measuredAt = canonicalDailyTimestamp(s.date); + // Fallback: no usable window. Anchor at local noon of the wake date and + // emit untimed stage rows (the pre-v1.17.1 shape) so a record missing the + // window fields still contributes a night total. + const measuredAt = canonicalDailyTimestamp(s.date, tz); if (Number.isNaN(measuredAt.getTime())) return []; for (const [sec, stage, fieldTag] of stages) { if (typeof sec === "number" && sec >= 0) { @@ -625,15 +633,19 @@ export function mapSleep(s: PolarSleep): MappedMeasurement[] { fieldTag: "sleep_score", }); } - return out; + return out.map((m) => ({ ...m, day: s.date })); } /** Map one current AccessLink daily-activity record: `steps` + * `active_calories` where Polar reports them. The active energy portion is * used instead of total `calories`, which includes BMR. */ -export function mapActivity(a: PolarActivity): MappedMeasurement[] { +export function mapActivity( + a: PolarActivity, + tz?: string, +): MappedMeasurement[] { if (typeof a.start_time !== "string") return []; - const measuredAt = canonicalDailyTimestamp(a.start_time.slice(0, 10)); + const day = a.start_time.slice(0, 10); + const measuredAt = canonicalDailyTimestamp(day, tz); if (Number.isNaN(measuredAt.getTime())) return []; const out: MappedMeasurement[] = []; const steps = a.steps; @@ -669,14 +681,17 @@ export function mapActivity(a: PolarActivity): MappedMeasurement[] { fieldTag: "distance", }); } - return out; + return out.map((m) => ({ ...m, day })); } /** Map one Polar Training Load Pro record: `cardio_load` → `CARDIO_LOAD` * (Polar's device-native cardiovascular-strain figure). `0` is a valid load, so * guard on `number` only. */ -export function mapCardioLoad(c: PolarCardioLoad): MappedMeasurement[] { - const measuredAt = canonicalDailyTimestamp(c.date); +export function mapCardioLoad( + c: PolarCardioLoad, + tz?: string, +): MappedMeasurement[] { + const measuredAt = canonicalDailyTimestamp(c.date, tz); if (Number.isNaN(measuredAt.getTime())) return []; if (typeof c.cardio_load !== "number" || c.cardio_load < 0) return []; return [ @@ -686,6 +701,7 @@ export function mapCardioLoad(c: PolarCardioLoad): MappedMeasurement[] { unit: "score", measuredAt, fieldTag: "cardio_load", + day: c.date, }, ]; } diff --git a/src/lib/polar/sync.ts b/src/lib/polar/sync.ts index f249f3288..db71ff320 100644 --- a/src/lib/polar/sync.ts +++ b/src/lib/polar/sync.ts @@ -81,6 +81,7 @@ import { } from "@/lib/sleep/sweep-stale-segments"; import { getPolarConnection } from "./credentials"; import { PolarApiError, classifyPolarError } from "./response-classifier"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; /** Map a Polar error onto the shared integration-ledger failure kind. */ export function classifyPolarFailure(err: unknown): FailureKind { @@ -134,12 +135,12 @@ function toUpsert( // Reconstructed sleep segments supply their own indexed externalId so the // several rows of one night stay distinct; everything else keys on // `::` — stable across re-syncs of the same day. - // The date is read off `measuredAt` for untimed rows (midnight-UTC - // anchored), so a mapper-supplied externalId is honoured verbatim to avoid - // the timed sleep instants drifting the date slice. + // A date-keyed row names its date (`day`): its instant is local noon in + // the user's zone, which east of UTC+12 falls on the previous UTC date. + // A timed row keys on its instant's UTC date, as it always has. externalId: m.externalId ?? - `${resourcePrefix}:${m.measuredAt.toISOString().slice(0, 10)}:${m.fieldTag}`, + `${resourcePrefix}:${m.day ?? m.measuredAt.toISOString().slice(0, 10)}:${m.fieldTag}`, sleepStage: m.sleepStage ?? null, })); } @@ -178,6 +179,8 @@ export interface PolarCollectionFailure { export async function syncUserPolar(userId: string): Promise { const conn = await getPolarConnection(userId); if (!conn) return { imported: 0, failed: false }; + // Date-keyed rows are anchored at local noon of their date in this zone. + const tz = await resolveUserTimezone(userId); // Filled by the sleep collection's closure as it maps records; carried so the // caller can run the record-scoped sweep before the upsert. A collection that @@ -190,7 +193,7 @@ export async function syncUserPolar(userId: string): Promise { name: "recharge", collect: async (t) => (await fetchNightlyRecharges(t)).flatMap((r) => - toUpsert(mapNightlyRecharge(r), "recharge"), + toUpsert(mapNightlyRecharge(r, tz), "recharge"), ), }, { @@ -199,7 +202,7 @@ export async function syncUserPolar(userId: string): Promise { const records = await fetchSleeps(t); const rowsOut: PolarMeasurementUpsert[] = []; for (const s of records) { - const rows = toUpsert(mapSleep(s), "sleep"); + const rows = toUpsert(mapSleep(s, tz), "sleep"); rowsOut.push(...rows); // Night-scoped sweep entry: the reconstructed segments of this date // all key under `sleep::seg:` (mapper-supplied). Any live row @@ -223,14 +226,14 @@ export async function syncUserPolar(userId: string): Promise { name: "activity", collect: async (t) => (await fetchActivities(t)).flatMap((a) => - toUpsert(mapActivity(a), "activity"), + toUpsert(mapActivity(a, tz), "activity"), ), }, { name: "cardioload", collect: async (t) => (await fetchCardioLoads(t)).flatMap((c) => - toUpsert(mapCardioLoad(c), "cardioload"), + toUpsert(mapCardioLoad(c, tz), "cardioload"), ), }, { diff --git a/src/lib/withings/__tests__/sync-activity.test.ts b/src/lib/withings/__tests__/sync-activity.test.ts index 8c1ba17fe..bd8c9db39 100644 --- a/src/lib/withings/__tests__/sync-activity.test.ts +++ b/src/lib/withings/__tests__/sync-activity.test.ts @@ -32,6 +32,10 @@ vi.mock("@/lib/db", () => { }; }); +vi.mock("@/lib/tz/resolver", () => ({ + resolveUserTimezone: vi.fn(), +})); + vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn().mockResolvedValue(false), parkIntegrationAtReauth: vi.fn(), @@ -78,6 +82,7 @@ import { recordSyncSuccess, } from "@/lib/integrations/status"; import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; import { fetchWithingsActivity, syncUserActivity } from "../sync-activity"; @@ -114,6 +119,7 @@ beforeEach(() => { vi.mocked(prisma.withingsConnection.findUnique).mockResolvedValue({ scope: "user.metrics,user.activity", } as never); + vi.mocked(resolveUserTimezone).mockResolvedValue("UTC"); }); afterEach(() => { @@ -253,6 +259,7 @@ describe("syncUserActivity — field mapping + idempotency", () => { id: "row-steps", type: "ACTIVITY_STEPS", measuredAt, + externalId: "withings:activity:user-1:2026-05-12:steps", value: 8420, deletedAt: null, }, @@ -296,38 +303,89 @@ describe("syncUserActivity — field mapping + idempotency", () => { expect(arg.data.deletedAt).toBeNull(); }); - it("anchors measuredAt at noon UTC so the instant lands inside the local day for every supported tz", async () => { - installFetchMock([{ date: "2026-05-12", steps: 1000 }]); - vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); - vi.mocked(prisma.measurement.createMany).mockResolvedValue({ - count: 1, - } as never); + it.each([ + ["UTC", "2026-05-12T12:00:00.000Z"], + ["America/New_York", "2026-05-12T16:00:00.000Z"], + ["Europe/Berlin", "2026-05-12T10:00:00.000Z"], + ["Pacific/Auckland", "2026-05-12T00:00:00.000Z"], + ["Pacific/Tongatapu", "2026-05-11T23:00:00.000Z"], + ])( + "anchors a %s day at local noon, which reads the same calendar day there", + async (tz, expected) => { + vi.mocked(resolveUserTimezone).mockResolvedValue(tz); + installFetchMock([{ date: "2026-05-12", steps: 1000 }]); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 1, + } as never); + + await syncUserActivity("user-1"); + + const measuredAt = createManyRows()[0].measuredAt as Date; + expect(measuredAt.toISOString()).toBe(expected); + expect( + new Intl.DateTimeFormat("en-CA", { + timeZone: tz, + year: "numeric", + month: "2-digit", + day: "2-digit", + }).format(measuredAt), + ).toBe("2026-05-12"); + }, + ); + + it("moves a row stored at the old noon-UTC anchor to local noon instead of adding a second one", async () => { + // Noon UTC is already 01:00 the next day in Tonga, so the old anchor + // counted 12 May's steps on 13 May there. + vi.mocked(resolveUserTimezone).mockResolvedValue("Pacific/Tongatapu"); + installFetchMock([{ date: "2026-05-12", steps: 8420 }]); + const legacy = new Date("2026-05-12T12:00:00.000Z"); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + id: "row-steps", + type: "ACTIVITY_STEPS", + measuredAt: legacy, + externalId: "withings:activity:user-1:2026-05-12:steps", + value: 8420, + deletedAt: null, + }, + ] as never); + vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); await syncUserActivity("user-1"); - const measuredAt = createManyRows()[0].measuredAt as Date; - expect(measuredAt.toISOString()).toBe("2026-05-12T12:00:00.000Z"); + expect(prisma.measurement.createMany).not.toHaveBeenCalled(); + expect(prisma.measurement.update).toHaveBeenCalledTimes(1); + const arg = vi.mocked(prisma.measurement.update).mock.calls[0][0] as { + data: { measuredAt: Date }; + }; + expect(arg.data.measuredAt.toISOString()).toBe("2026-05-11T23:00:00.000Z"); + // Both the day the row left and the day it joined are refolded. + const refolded = vi + .mocked(recomputeBucketsForMeasurement) + .mock.calls.map((c) => (c[2] as Date).toISOString().slice(0, 10)); + expect(refolded).toEqual( + expect.arrayContaining(["2026-05-11", "2026-05-12"]), + ); + }); - // Regression: anchoring at noon UTC keeps the row inside the - // calendar day every user reads it in. Bucketing the same instant - // via `Intl.DateTimeFormat` across the canonical Withings user - // span — Honolulu (UTC-10), Los Angeles (UTC-7/-8), Berlin - // (UTC+1/+2) and Tokyo (UTC+9) — must all return "2026-05-12". - // Anchoring at end-of-day UTC (the v1.4.25 W17b shape) shifted - // Tokyo by +1 day; anchoring at midnight UTC would shift LA by -1. - // Noon UTC is the only choice that holds across the practical - // [-11, +12) range. - const bucket = (tz: string) => - new Intl.DateTimeFormat("en-CA", { - timeZone: tz, - year: "numeric", - month: "2-digit", - day: "2-digit", - }).format(measuredAt); - expect(bucket("Pacific/Honolulu")).toBe("2026-05-12"); - expect(bucket("America/Los_Angeles")).toBe("2026-05-12"); - expect(bucket("Europe/Berlin")).toBe("2026-05-12"); - expect(bucket("Asia/Tokyo")).toBe("2026-05-12"); + it("fetches through today's date in the user's zone", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-05-12T20:00:00.000Z")); + try { + vi.mocked(resolveUserTimezone).mockResolvedValue("Pacific/Auckland"); + const fetchMock = installFetchMock([]); + await syncUserActivity("user-1"); + const [, init] = fetchMock.mock.calls[0] as unknown as [ + string, + { body: unknown }, + ]; + const body = String(init.body); + // 08:00 on 13 May in Auckland: the window ends on the 13th. + expect(body).toContain("enddateymd=2026-05-13"); + } finally { + vi.useRealTimers(); + } }); it("skips missing fields without writing zero rows or throwing", async () => { @@ -440,10 +498,18 @@ describe("syncUserActivity — field mapping + idempotency", () => { // The single read is keyed by the full distinct (type, measuredAt) // sets the batch will write — `skipDuplicates` guards the insert. const readArg = vi.mocked(prisma.measurement.findMany).mock.calls[0][0] as { - where: { type: { in: unknown[] }; measuredAt: { in: unknown[] } }; + where: { + type: { in: unknown[] }; + OR: [ + { externalId: { in: unknown[] } }, + { measuredAt: { in: unknown[] } }, + ]; + }; }; expect(readArg.where.type.in).toHaveLength(3); - expect(readArg.where.measuredAt.in).toHaveLength(5); + expect(readArg.where.OR[0].externalId.in).toHaveLength(15); + // UTC: the local-noon and the old noon-UTC anchor coincide. + expect(readArg.where.OR[1].measuredAt.in).toHaveLength(5); const createArg = vi.mocked(prisma.measurement.createMany).mock .calls[0][0] as { skipDuplicates: boolean }; expect(createArg.skipDuplicates).toBe(true); diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index a4a0098b2..18ca9c4b5 100644 --- a/src/lib/withings/sync-activity.ts +++ b/src/lib/withings/sync-activity.ts @@ -36,6 +36,9 @@ import type { MeasurementType } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; +import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; +import { shiftDateKey, userDayKey } from "@/lib/tz/format"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; import { getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; import { getUnitForType } from "@/lib/validations/measurement"; @@ -112,16 +115,11 @@ const ACTIVITY_FIELD_MAP: ReadonlyArray<{ ]; /** - * Anchor a per-day activity row at noon UTC. Chosen so the instant - * always lands inside the local day for users in the [-11, +12] zone - * range — anchoring at end-of-day UTC (the v1.4.25 W17b shape) sent - * Tokyo readings into the *following* local day, mis-bucketing every - * positive-offset user's "today" tile. Noon UTC is the standard - * "calendar-day with no clock" representation per RFC 3339 §5.6 and - * the JS `Date` analytics layer day-keys it cleanly in every - * supported user timezone. + * Where the per-day activity rows sat up to v1.39.2: noon UTC of the date. + * Still read, so a row written under that anchor is found and moved to the + * local-noon anchor instead of being duplicated. */ -function activityMeasuredAt(yyyymmdd: string): Date { +function legacyActivityMeasuredAt(yyyymmdd: string): Date { return new Date(`${yyyymmdd}T12:00:00.000Z`); } @@ -268,18 +266,18 @@ export async function syncUserActivity( // 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); + // Withings activity is keyed by the user's calendar date, so the window + // is cut on the user's own days: a UTC cut left "today" out of the fetch + // until UTC caught up with a zone ahead of it. + const timeZone = await resolveUserTimezone(userId); + const today = userDayKey(new Date(), timeZone); let entries: WithingsActivityEntry[]; try { entries = await fetchWithingsActivity( tokenInfo.accessToken, - ymd(start), - ymd(now), + shiftDateKey(today, -ACTIVITY_BACKFILL_DAYS), + today, ); } catch (err) { // v1.4.43 W7-B3 — typed-classification path. The thrown @@ -340,13 +338,19 @@ export async function syncUserActivity( interface PlannedRow { type: MeasurementType; measuredAt: Date; + /** The pre-v1.39.3 noon-UTC anchor of the same date. */ + legacyMeasuredAt: Date; value: number; externalId: string; } const collected: PlannedRow[] = []; for (const entry of entries) { if (!entry.date) continue; - const measuredAt = activityMeasuredAt(entry.date); + // Local noon of the date in the user's zone: the instant reads the same + // calendar day there. Noon UTC read as the next day from UTC+13 on + // (Auckland in summer, Tonga, Samoa), so every daily total landed a day + // late for those users. + const measuredAt = canonicalDailyTimestamp(entry.date, timeZone); for (const { field, type } of ACTIVITY_FIELD_MAP) { const raw = entry[field]; // Withings returns 0 for "active but no movement" — that's @@ -356,6 +360,7 @@ export async function syncUserActivity( collected.push({ type, measuredAt, + legacyMeasuredAt: legacyActivityMeasuredAt(entry.date), value: raw, externalId: `withings:activity:${userId}:${entry.date}:${field}`, }); @@ -377,13 +382,11 @@ export async function syncUserActivity( try { // One read for the whole batch: every existing WITHINGS / non-sleep - // row that occupies a slot we intend to write. Keyed by the same - // four columns the per-row `findFirst` matched on, so the - // overwrite-vs-insert decision is identical — just resolved in - // memory instead of N round-trips. - const distinctMeasuredAt = [ - ...new Map(planned.map((p) => [p.measuredAt.getTime(), p.measuredAt])), - ].map(([, d]) => d); + // row for a planned day, found by its stable `externalId` or, for a + // row written before it carried one, by the day's old noon-UTC slot. + // A match is rewritten in place, including its `measuredAt`, so a row + // written under the old anchor moves to local noon on the next sync + // rather than being duplicated beside it. const distinctTypes = [...new Set(planned.map((p) => p.type))]; const existingRows = await prisma.measurement.findMany({ where: { @@ -391,40 +394,65 @@ export async function syncUserActivity( source: "WITHINGS", sleepStage: null, type: { in: distinctTypes }, - measuredAt: { in: distinctMeasuredAt }, + OR: [ + { externalId: { in: planned.map((p) => p.externalId) } }, + { + measuredAt: { + in: [ + ...new Map( + planned.flatMap((p) => [ + [p.measuredAt.getTime(), p.measuredAt], + [p.legacyMeasuredAt.getTime(), p.legacyMeasuredAt], + ]), + ).values(), + ], + }, + }, + ], }, select: { id: true, type: true, measuredAt: true, + externalId: true, value: true, deletedAt: true, }, }); - const existingBySlot = new Map< - string, - { id: string; value: number; deletedAt: Date | null } - >(); + type Existing = (typeof existingRows)[number]; + const byExternalId = new Map(); + const bySlot = new Map(); for (const row of existingRows) { - existingBySlot.set(slotKey(row.type, row.measuredAt), { - id: row.id, - value: row.value, - deletedAt: row.deletedAt, - }); + if (row.externalId) byExternalId.set(row.externalId, row); + bySlot.set(slotKey(row.type, row.measuredAt), row); } const toCreate: PlannedRow[] = []; - const toUpdate: Array<{ id: string; value: number }> = []; + const toUpdate: Array<{ id: string; p: PlannedRow }> = []; for (const p of planned) { - const existing = existingBySlot.get(slotKey(p.type, p.measuredAt)); + const existing = + byExternalId.get(p.externalId) ?? + bySlot.get(slotKey(p.type, p.measuredAt)) ?? + bySlot.get(slotKey(p.type, p.legacyMeasuredAt)); if (existing) { - // Only write when the value actually moved — a re-sync of an - // unchanged LIVE day is then a pure no-op on the write path. A - // TOMBSTONED row always writes: the update is what resurrects it - // (`deletedAt: null`) — Withings is the source of truth for its - // own rows (mirrors Google / Fitbit). - if (existing.value !== p.value || existing.deletedAt !== null) { - toUpdate.push({ id: existing.id, value: p.value }); + // Only write when something moved — a re-sync of an unchanged LIVE + // day is then a pure no-op on the write path. A TOMBSTONED row + // always writes: the update is what resurrects it (`deletedAt: + // null`) — Withings is the source of truth for its own rows + // (mirrors Google / Fitbit). + const moved = + existing.measuredAt.getTime() !== p.measuredAt.getTime(); + if ( + existing.value !== p.value || + existing.deletedAt !== null || + moved || + existing.externalId !== p.externalId + ) { + toUpdate.push({ id: existing.id, p }); + } + // The day the row leaves needs its rollup refolded too. + if (moved) { + touched.push({ type: p.type, measuredAt: existing.measuredAt }); } } else { toCreate.push(p); @@ -461,7 +489,12 @@ export async function syncUserActivity( where: { id: u.id }, // `deletedAt: null` — no-op on a live row, deliberate // RESURRECTION on a tombstoned one (see the skip above). - data: { value: u.value, deletedAt: null }, + data: { + value: u.p.value, + measuredAt: u.p.measuredAt, + externalId: u.p.externalId, + deletedAt: null, + }, }), ), );