diff --git a/src/lib/fitbit/__tests__/client.test.ts b/src/lib/fitbit/__tests__/client.test.ts index 23bbf4081..483bf42d5 100644 --- a/src/lib/fitbit/__tests__/client.test.ts +++ b/src/lib/fitbit/__tests__/client.test.ts @@ -503,6 +503,62 @@ describe("sleep mapping (1.2 levels.data)", () => { expect(mapSleepSession({ levels: { data: [] } })).toHaveLength(0); expect(mapSleepSession({})).toHaveLength(0); }); + + it("anchors a near-midnight segment END to the user's timezone, not the process zone", () => { + // A Berlin user (UTC+2 in May / CEST) whose last segment ENDS at 00:30 + // local on 2026-05-11. The offset-less wall clock `2026-05-11T00:30:00` + // must resolve to 22:30 UTC on 2026-05-10 — NOT 00:30 UTC (which a bare + // `new Date(iso)` in a UTC process would produce, flipping the wake-day). + const sessions = readSleepSessions({ + sleep: [ + { + logId: 4242, + startTime: "2026-05-10T23:00:00.000", + endTime: "2026-05-11T00:30:00.000", + levels: { + data: [ + { + dateTime: "2026-05-11T00:00:00.000", + level: "deep", + seconds: 1800, // ends 00:30 Berlin local + }, + ], + }, + }, + ], + }); + + const out = mapSleepSession(sessions[0]!, "Europe/Berlin"); + expect(out).toHaveLength(1); + // 00:30 CEST (UTC+2) → 22:30 UTC on the PRIOR civil day. + expect(out[0]!.measuredAt.toISOString()).toBe("2026-05-10T22:30:00.000Z"); + }); + + it("anchors a logId-less session anchor against the user's timezone", () => { + const sessions = readSleepSessions({ + sleep: [ + { + // No logId → the anchor falls back to the END instant, which must + // also be tz-resolved so the externalId is stable across syncs. + endTime: "2026-05-11T00:30:00.000", + levels: { + data: [ + { + dateTime: "2026-05-11T00:00:00.000", + level: "rem", + seconds: 1800, + }, + ], + }, + }, + ], + }); + + const out = mapSleepSession(sessions[0]!, "Europe/Berlin"); + expect(out).toHaveLength(1); + // anchor = end instant ISO (UTC) → fieldTag starts with that instant. + expect(out[0]!.fieldTag).toBe("2026-05-10T22:30:00.000Z:sleep_rem:0"); + }); }); describe("workout mapping (activities list)", () => { diff --git a/src/lib/fitbit/__tests__/sync.test.ts b/src/lib/fitbit/__tests__/sync.test.ts index 58be146a8..6d4047964 100644 --- a/src/lib/fitbit/__tests__/sync.test.ts +++ b/src/lib/fitbit/__tests__/sync.test.ts @@ -14,6 +14,7 @@ const { prismaMock, refreshAccessTokenMock, recordSyncFailure } = vi.hoisted( fitbitConnection: { findUnique: vi.fn(), update: vi.fn(), + updateMany: vi.fn(), }, measurement: { findMany: vi.fn<(arg: Record) => Promise>( @@ -103,7 +104,7 @@ describe("getValidToken — rotating refresh", () => { access_token: "new-access", expires_in: 3600, }); - prismaMock.fitbitConnection.update.mockResolvedValue({}); + prismaMock.fitbitConnection.updateMany.mockResolvedValue({ count: 1 }); const result = await getValidToken("user1"); @@ -112,7 +113,7 @@ describe("getValidToken — rotating refresh", () => { clientId: "cid", clientSecret: "csecret", }); - const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0]; + const updateArg = prismaMock.fitbitConnection.updateMany.mock.calls[0]![0]; expect(updateArg.data.accessToken).toBe("enc(new-access)"); // The stored refresh token must NOT be wiped when the response omits one. expect(updateArg.data).not.toHaveProperty("refreshToken"); @@ -131,15 +132,42 @@ describe("getValidToken — rotating refresh", () => { refresh_token: "rotated-refresh", expires_in: 3600, }); - prismaMock.fitbitConnection.update.mockResolvedValue({}); + prismaMock.fitbitConnection.updateMany.mockResolvedValue({ count: 1 }); await getValidToken("user1"); - const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0]; + const updateArg = prismaMock.fitbitConnection.updateMany.mock.calls[0]![0]; + // CAS guard scopes the write to the connection AND the spent ciphertext. + expect(updateArg.where.id).toBe("conn1"); + expect(updateArg.where.refreshToken).toBe("enc(old-refresh)"); expect(updateArg.data.accessToken).toBe("enc(new-access)"); expect(updateArg.data.refreshToken).toBe("enc(rotated-refresh)"); }); + it("reuses the peer's rotated token on a lost CAS race (no spurious reauth)", async () => { + prismaMock.fitbitConnection.findUnique + .mockResolvedValueOnce({ + id: "conn1", + fitbitUserId: "abc123", + accessToken: "enc(old-access)", + refreshToken: "enc(old-refresh)", + tokenExpiresAt: new Date(Date.now() - 1000), + }) + .mockResolvedValueOnce({ accessToken: "enc(peer-access)" }); + refreshAccessTokenMock.mockResolvedValue({ + access_token: "new-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + // A concurrent sync rotated first → zero rows match the CAS guard. + prismaMock.fitbitConnection.updateMany.mockResolvedValue({ count: 0 }); + + const result = await getValidToken("user1"); + + expect(result?.accessToken).toBe("peer-access"); + expect(recordSyncFailure).not.toHaveBeenCalled(); + }); + it("returns the stored token without refresh when not near expiry", async () => { prismaMock.fitbitConnection.findUnique.mockResolvedValue({ id: "conn1", diff --git a/src/lib/fitbit/client.ts b/src/lib/fitbit/client.ts index 22a59049f..e5eab0e07 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 { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; import { FitbitApiError, classifyFitbitResponse } from "./response-classifier"; /** Classic Fitbit Web API base. */ @@ -929,13 +930,13 @@ interface FitbitSleepSession { * The stable session anchor for a sleep externalId — the logId, else the end (or * start) instant. A re-scored night reuses the same logId → overwrites in place. */ -function sleepSessionAnchor(s: FitbitSleepSession): string { +function sleepSessionAnchor(s: FitbitSleepSession, tz?: string): string { if (typeof s.logId === "number" && Number.isFinite(s.logId)) { return String(s.logId); } for (const t of [s.endTime, s.startTime]) { if (typeof t === "string") { - const d = parseLocalInstant(t); + const d = parseLocalInstant(t, tz); if (d) return d.toISOString(); } } @@ -943,12 +944,40 @@ function sleepSessionAnchor(s: FitbitSleepSession): string { } /** - * Parse a Fitbit sleep timestamp. The 1.2 sleep log emits LOCAL wall-clock ISO - * strings WITHOUT an offset (e.g. `2020-01-26T03:02:30.000`); `new Date(...)` - * treats an offset-less ISO string as local time, which is the user's intent - * here (the night belongs to the user's local clock). Returns null on a miss. + * Matches an offset-less local ISO wall-clock string Fitbit emits on the classic + * sleep + activity logs (`2020-01-26T03:02:30` / `...T03:02:30.000`). No trailing + * `Z`, no `±hh:mm` — those denote an absolute instant and are honoured verbatim. */ -function parseLocalInstant(iso: string): Date | null { +const OFFSET_LESS_LOCAL_ISO = + /^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2})(?::(\d{2}))?(?:\.\d+)?$/; + +/** + * Parse a Fitbit timestamp into a UTC instant. The classic 1.2 sleep log and the + * activities list emit LOCAL wall-clock strings WITHOUT an offset (e.g. + * `2020-01-26T03:02:30.000`) — the night/activity belongs to the user's local + * clock, so the wall clock must be resolved against the USER'S timezone, not the + * process zone. A bare `new Date(iso)` parses an offset-less string in the host + * zone (UTC in production), which shifts a non-UTC user's segment by their UTC + * offset and can flip the wake-day in the night reconstruction. When `tz` is + * omitted (no user zone on the path) the host-local fallback preserves the prior + * behaviour. Strings that DO carry an offset/`Z` are absolute and parsed as-is. + * Returns null on a miss. + */ +function parseLocalInstant(iso: string, tz?: string): Date | null { + const m = OFFSET_LESS_LOCAL_ISO.exec(iso.trim()); + if (m) { + return zonedWallClockToUtc( + { + year: Number(m[1]), + month: Number(m[2]), + day: Number(m[3]), + hour: Number(m[4]), + minute: Number(m[5]), + second: m[6] ? Number(m[6]) : 0, + }, + tz, + ); + } const d = new Date(iso); return Number.isNaN(d.getTime()) ? null : d; } @@ -959,14 +988,20 @@ function parseLocalInstant(iso: string): Date | null { * Each segment carries a stage-scoped, INDEXED fieldTag so the several segments * of one stage stay distinct under the `(userId, type, source, externalId)` * dedup key. Unknown stage labels are skipped. + * + * The 1.2 sleep log emits OFFSET-LESS local wall-clock timestamps, so `tz` (the + * user's stored zone) anchors them to the correct UTC instant — without it a + * non-UTC user's near-midnight segment END would shift by their UTC offset and + * could land on the wrong wake-day in the night reconstruction. */ export function mapSleepSession( session: FitbitSleepSession, + tz?: string, ): FitbitMappedMeasurement[] { const data = session.levels?.data; if (!Array.isArray(data) || data.length === 0) return []; - const anchor = sleepSessionAnchor(session); + const anchor = sleepSessionAnchor(session, tz); const out: FitbitMappedMeasurement[] = []; let segIndex = 0; for (const seg of data) { @@ -975,7 +1010,7 @@ export function mapSleepSession( if (!positive(seg.seconds)) continue; const startStr = typeof seg.dateTime === "string" ? seg.dateTime : null; if (!startStr) continue; - const start = parseLocalInstant(startStr); + const start = parseLocalInstant(startStr, tz); if (!start) continue; const minutes = seg.seconds / 60; const end = new Date(start.getTime() + seg.seconds * 1000); @@ -1083,13 +1118,18 @@ export interface FitbitMappedWorkout { * there is no usable start + positive duration. The externalId anchors on the * `logId` (stable) so a re-fetch overwrites the same row. The classic * activities-list endpoint does not surface min/max HR, so those stay null. + * + * `startTime` is an offset-less local wall-clock string, so `tz` (the user's + * stored zone) anchors it to the correct UTC instant rather than the process + * zone. */ export function mapWorkout( entry: FitbitActivityLogEntry, + tz?: string, ): FitbitMappedWorkout | null { const start = typeof entry.startTime === "string" - ? parseLocalInstant(entry.startTime) + ? parseLocalInstant(entry.startTime, tz) : null; if (!start) return null; if (!positive(entry.duration)) return null; diff --git a/src/lib/fitbit/sync-sleep.ts b/src/lib/fitbit/sync-sleep.ts index 2215d4ef8..662cfbc46 100644 --- a/src/lib/fitbit/sync-sleep.ts +++ b/src/lib/fitbit/sync-sleep.ts @@ -32,6 +32,7 @@ import { type FitbitResourceSyncOptions, } from "./sync"; import { annotate } from "@/lib/logging/context"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; export async function syncUserSleep( userId: string, @@ -40,6 +41,11 @@ export async function syncUserSleep( const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; + // The classic 1.2 sleep log emits offset-less local wall-clock timestamps; + // anchor them against the user's stored zone so a near-midnight segment END + // lands on the correct wake-day rather than being shifted by the process zone. + 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 windows = chunkDateRanges(start, end, FITBIT_SLEEP_RANGE_DAYS); @@ -53,7 +59,7 @@ export async function syncUserSleep( return handleCollectionFetchError("fetchSleep", userId, err); } for (const session of readSleepSessions(body)) { - for (const m of mapSleepSession(session)) { + for (const m of mapSleepSession(session, tz)) { readings.push({ type: m.type, value: m.value, diff --git a/src/lib/fitbit/sync-workout.ts b/src/lib/fitbit/sync-workout.ts index 7e38a3872..a430e85f7 100644 --- a/src/lib/fitbit/sync-workout.ts +++ b/src/lib/fitbit/sync-workout.ts @@ -29,6 +29,7 @@ import { } from "./sync"; import { prisma } from "@/lib/db"; import { annotate, getEvent } from "@/lib/logging/context"; +import { resolveUserTimezone } from "@/lib/tz/resolver"; /** Activities per page on the `activities/list` walk. */ const WORKOUT_PAGE_SIZE = 100; @@ -42,6 +43,10 @@ export async function syncUserWorkout( const tokenInfo = await getValidToken(userId); if (!tokenInfo) return 0; + // `startTime` on the activities list is offset-less local wall-clock; anchor + // it to the user's stored zone rather than the process zone. + const tz = await resolveUserTimezone(userId); + const afterDate = opts.start ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); @@ -67,7 +72,7 @@ export async function syncUserWorkout( let imported = 0; for (const entry of entries) { - const w: FitbitMappedWorkout | null = mapWorkout(entry); + const w: FitbitMappedWorkout | null = mapWorkout(entry, tz); if (!w) continue; // no usable time span — not a workout try { diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index f85eda2ad..6abf49dcd 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -5,7 +5,9 @@ * `tokenExpiresAt - 5 min`, and persists the new access token + expiry AND * the rotated refresh token. Classic Fitbit ROTATES refresh tokens * (one-time use), so the new `refresh_token` MUST replace the stored one — - * the inverse of the prior Google transport, which did not rotate. + * the inverse of the prior Google transport, which did not rotate. The + * persist is a compare-and-swap on the stored ciphertext so two overlapping + * syncs cannot park the connection at reauth (see `persistRotatedToken`). * - `recordFitbitSyncFailure` / `classificationToFailureKind` map a * classified API error onto the shared integration-status ledger. * - `upsertFitbitMeasurements` + `syncUserFitbit` own the write tail (dedup, @@ -33,6 +35,7 @@ import { recomputeUserRollups, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { persistRotatedToken } from "@/lib/integrations/oauth-refresh"; import { refreshAccessToken } from "./client"; import { getUserFitbitCredentials } from "./credentials"; import { @@ -115,22 +118,47 @@ export async function getValidToken( // Classic Fitbit ROTATES refresh tokens: the refresh response carries a // fresh `refresh_token` that replaces the (now-invalid) stored one. Persist - // it unconditionally. Guard the malformed-reply case: if the response omits - // the new refresh token, keep the existing one rather than writing a blank, - // which would brick every future refresh. - await prisma.fitbitConnection.update({ - where: { id: connection.id }, - data: { - accessToken: encrypt(newTokens.access_token), - tokenExpiresAt: expiresAt, - ...(newTokens.refresh_token - ? { refreshToken: encrypt(newTokens.refresh_token) } - : {}), + // with a compare-and-swap on the stored ciphertext so a concurrent sync + // that already rotated the token does not get its work clobbered — and so + // THIS caller, if it lost the race, reuses the peer's rotated access token + // instead of parking the connection at reauth. Guard the malformed-reply + // case: if the response omits the new refresh token, keep the existing one + // rather than writing a blank, which would brick every future refresh. + const persistedAccessToken = await persistRotatedToken( + newTokens.access_token, + { + conditionalUpdate: async () => { + const { count } = await prisma.fitbitConnection.updateMany({ + // `connection.refreshToken` is the exact ciphertext we read + + // spent; matching it is the CAS guard (a re-encrypt would differ). + where: { + id: connection.id, + refreshToken: connection.refreshToken, + }, + data: { + accessToken: encrypt(newTokens.access_token), + tokenExpiresAt: expiresAt, + ...(newTokens.refresh_token + ? { refreshToken: encrypt(newTokens.refresh_token) } + : {}), + }, + }); + return count; + }, + readPeerAccessToken: async () => { + const fresh = await prisma.fitbitConnection.findUnique({ + where: { id: connection.id }, + select: { accessToken: true }, + }); + return fresh ? decrypt(fresh.accessToken) : null; + }, }, - }); + ); + + if (!persistedAccessToken) return null; return { - accessToken: newTokens.access_token, + accessToken: persistedAccessToken, connection: { id: connection.id, fitbitUserId: connection.fitbitUserId, diff --git a/src/lib/integrations/__tests__/oauth-refresh.test.ts b/src/lib/integrations/__tests__/oauth-refresh.test.ts new file mode 100644 index 000000000..8f7185107 --- /dev/null +++ b/src/lib/integrations/__tests__/oauth-refresh.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it, vi } from "vitest"; +import { persistRotatedToken } from "../oauth-refresh"; + +describe("persistRotatedToken", () => { + it("returns the freshly minted access token when the CAS write wins (1 row)", async () => { + const conditionalUpdate = vi.fn(async () => 1); + const readPeerAccessToken = vi.fn(async () => "peer"); + + const token = await persistRotatedToken("mine", { + conditionalUpdate, + readPeerAccessToken, + }); + + expect(token).toBe("mine"); + expect(conditionalUpdate).toHaveBeenCalledTimes(1); + // No re-read on a win. + expect(readPeerAccessToken).not.toHaveBeenCalled(); + }); + + it("reuses the peer's rotated token on a lost race (0 rows)", async () => { + const conditionalUpdate = vi.fn(async () => 0); + const readPeerAccessToken = vi.fn(async () => "peer"); + + const token = await persistRotatedToken("mine", { + conditionalUpdate, + readPeerAccessToken, + }); + + expect(token).toBe("peer"); + expect(readPeerAccessToken).toHaveBeenCalledTimes(1); + }); + + it("returns null when the row vanished after a lost race", async () => { + const token = await persistRotatedToken("mine", { + conditionalUpdate: async () => 0, + readPeerAccessToken: async () => null, + }); + + expect(token).toBeNull(); + }); +}); diff --git a/src/lib/integrations/oauth-refresh.ts b/src/lib/integrations/oauth-refresh.ts new file mode 100644 index 000000000..2298724d8 --- /dev/null +++ b/src/lib/integrations/oauth-refresh.ts @@ -0,0 +1,69 @@ +/** + * Shared OAuth rotating-refresh-token persistence helper. + * + * Fitbit, WHOOP, and Oura all issue ONE-TIME-USE refresh tokens: every refresh + * response carries a fresh `refresh_token` that invalidates the one just spent. + * The naive read → refresh → persist sequence is racy. When two syncs for the + * same user overlap (a scheduled tick collides with a manual sync, or two + * resource syncs each call `getValidToken`), both read the SAME stored refresh + * token, both POST it, and the provider honours only the first. The loser's POST + * comes back 401, the caller classifies it as `reauth_required`, and the + * connection is parked at `error_reauth` — spuriously, because the peer already + * rotated the token successfully and the connection is perfectly healthy. + * + * The fix is a compare-and-swap on persistence: write the rotated tokens with a + * `WHERE refreshToken = ` guard. The tokens are + * AES-256-GCM encrypted with a random IV, so the guard matches the STORED + * ciphertext (passed in verbatim), never a re-encryption of the plaintext — + * re-encrypting the same plaintext yields different bytes. If the conditional + * update touches zero rows a peer refresh already won; re-read the row and reuse + * the peer's freshly-rotated access token instead of erroring. + * + * Both racing callers thus end up with a usable access token and neither parks + * the connection. The window is small but real on the shared scheduler, and a + * single spurious reauth forces the user back through the consent flow. + */ + +/** Tokens returned by a provider's refresh endpoint (the fields we persist). */ +export interface RotatedTokens { + accessToken: string; + /** The new refresh token. Omitted only on a malformed reply (kept as-is). */ + refreshToken?: string | null; + /** Absolute expiry of the new access token, if the provider supplies one. */ + expiresAt?: Date; +} + +export interface PersistRotatedTokenOptions { + /** + * Conditional persist: update the connection row ONLY when its stored refresh + * token still equals the ciphertext this caller read (compare-and-swap on the + * stored bytes). Returns the number of rows written — 0 means a peer rotated + * first. The closure owns the encryption + the `WHERE` clause so the helper + * stays storage-agnostic across the three provider tables. + */ + conditionalUpdate: () => Promise; + /** + * Re-read the connection AFTER a lost race and return the peer's freshly + * rotated, DECRYPTED access token. Returns null only if the row vanished + * (connection deleted mid-flight) — the caller then treats it as no token. + */ + readPeerAccessToken: () => Promise; +} + +/** + * Persist a rotated token pair with compare-and-swap semantics, falling back to + * the peer's token on a lost race. Returns the access token the caller should + * use for the current sync — the one it just minted on a win, or the peer's on a + * loss — or null when the connection no longer exists. + */ +export async function persistRotatedToken( + freshAccessToken: string, + opts: PersistRotatedTokenOptions, +): Promise { + const rows = await opts.conditionalUpdate(); + if (rows > 0) return freshAccessToken; + // Lost the race: a concurrent refresh already rotated the stored token, which + // invalidated the one we spent. Reuse the peer's freshly rotated access token + // rather than parking the connection at reauth. + return opts.readPeerAccessToken(); +} diff --git a/src/lib/oura/__tests__/sync.test.ts b/src/lib/oura/__tests__/sync.test.ts index 52cd7742a..e452322f1 100644 --- a/src/lib/oura/__tests__/sync.test.ts +++ b/src/lib/oura/__tests__/sync.test.ts @@ -83,11 +83,17 @@ vi.mock("../client", async (importOriginal) => { import { syncUserOura } from "../sync"; import { OuraApiError } from "../response-classifier"; -const CONN = { accessToken: "acc", refreshToken: "ref" }; +const CONN = { + accessToken: "acc", + refreshToken: "ref", + refreshTokenCiphertext: "enc(ref)", +}; beforeEach(() => { getConnMock.mockReset(); - storeTokensMock.mockReset().mockResolvedValue(undefined); + // The compare-and-swap persist returns the access token the caller should use + // (its own on a CAS win) — default to the freshly minted one. + storeTokensMock.mockReset().mockResolvedValue("newAcc"); getCredsMock .mockReset() .mockResolvedValue({ clientId: "c", clientSecret: "s" }); @@ -155,11 +161,40 @@ describe("syncUserOura", () => { }); const imported = await syncUserOura("u1"); expect(refreshMock).toHaveBeenCalledWith("ref", expect.anything()); - expect(storeTokensMock).toHaveBeenCalledWith("u1", "newAcc", "newRef"); + expect(storeTokensMock).toHaveBeenCalledWith( + "u1", + "newAcc", + "newRef", + "enc(ref)", + ); expect(imported).toBe(1); expect(recordSuccessMock).toHaveBeenCalledWith("u1", "oura"); }); + it("reuses the peer's rotated token on a lost CAS race and still completes (no reauth)", async () => { + getConnMock.mockResolvedValue(CONN); + fetchReadinessMock + .mockRejectedValueOnce(err401()) + .mockResolvedValueOnce([{ id: "x", day: "2026-06-10", score: 70 }]); + refreshMock.mockResolvedValue({ + access_token: "newAcc", + refresh_token: "newRef", + expires_in: 86400, + }); + // A concurrent sync rotated first → the persist returns the PEER's token. + storeTokensMock.mockResolvedValue("peerAcc"); + + const imported = await syncUserOura("u1"); + + // The retry runs with the peer's token rather than the invalidated one. + expect(fetchReadinessMock).toHaveBeenLastCalledWith( + "peerAcc", + expect.anything(), + ); + expect(imported).toBe(1); + expect(recordFailureMock).not.toHaveBeenCalled(); + }); + it("records reauth_required when the refresh itself fails", async () => { getConnMock.mockResolvedValue(CONN); fetchReadinessMock.mockRejectedValue(err401()); diff --git a/src/lib/oura/credentials.ts b/src/lib/oura/credentials.ts index 32e9df380..bcf74f3a6 100644 --- a/src/lib/oura/credentials.ts +++ b/src/lib/oura/credentials.ts @@ -12,6 +12,7 @@ */ import { prisma } from "@/lib/db"; import { decrypt, encrypt } from "@/lib/crypto"; +import { persistRotatedToken } from "@/lib/integrations/oauth-refresh"; import { getOuraCredentials, type OuraCredentials } from "./client"; /** @@ -77,6 +78,12 @@ export async function clearOuraClientCredentials( export interface OuraConnection { accessToken: string; refreshToken: string; + /** + * The exact stored refresh-token ciphertext (NOT the decrypted value). The + * rotating-refresh compare-and-swap in `storeOuraTokens` matches on this so a + * concurrent sync that already rotated the token is not clobbered. + */ + refreshTokenCiphertext: string; } export async function getOuraConnection( @@ -95,20 +102,49 @@ export async function getOuraConnection( return { accessToken: decrypt(user.ouraAccessTokenEncrypted), refreshToken: decrypt(user.ouraRefreshTokenEncrypted), + refreshTokenCiphertext: user.ouraRefreshTokenEncrypted, }; } -/** Persist a rotated token pair after a successful refresh. */ +/** + * Persist a rotated Oura token pair with compare-and-swap on the stored refresh + * ciphertext (the inverse of Fitbit / WHOOP, which key by connection id). Oura's + * tokens live on `User` and the refresh is REACTIVE on a 401, so two overlapping + * syncs can race the same one-time-use refresh token. The CAS guard writes only + * when the stored ciphertext still equals the one this caller spent; on a lost + * race (0 rows) it re-reads and returns the peer's freshly rotated access token. + * + * Returns the access token the caller should use for the current sync (its own + * on a win, the peer's on a loss), or null if the connection vanished. + */ export async function storeOuraTokens( userId: string, accessToken: string, refreshToken: string, -): Promise { - await prisma.user.update({ - where: { id: userId }, - data: { - ouraAccessTokenEncrypted: encrypt(accessToken), - ouraRefreshTokenEncrypted: encrypt(refreshToken), + expectedRefreshCiphertext: string, +): Promise { + return persistRotatedToken(accessToken, { + conditionalUpdate: async () => { + const { count } = await prisma.user.updateMany({ + where: { + id: userId, + ouraRefreshTokenEncrypted: expectedRefreshCiphertext, + }, + data: { + ouraAccessTokenEncrypted: encrypt(accessToken), + ouraRefreshTokenEncrypted: encrypt(refreshToken), + }, + }); + return count; + }, + readPeerAccessToken: async () => { + const fresh = await prisma.user.findUnique({ + where: { id: userId }, + select: { ouraAccessTokenEncrypted: true }, + }); + return fresh?.ouraAccessTokenEncrypted + ? decrypt(fresh.ouraAccessTokenEncrypted) + : null; }, }); } diff --git a/src/lib/oura/sync.ts b/src/lib/oura/sync.ts index cefddf77b..363b3c485 100644 --- a/src/lib/oura/sync.ts +++ b/src/lib/oura/sync.ts @@ -124,6 +124,7 @@ async function fetchAll( userId: string, accessToken: string, refreshToken: string, + refreshTokenCiphertext: string, lookbackDays: number, ): Promise { const now = new Date(); @@ -179,8 +180,17 @@ async function fetchAll( if (!creds) throw err; const rotated = await refreshAccessToken(refreshToken, creds); - await storeOuraTokens(userId, rotated.access_token, rotated.refresh_token); - return run(rotated.access_token); + // Compare-and-swap persist: on a lost race against a concurrent sync this + // returns the peer's freshly rotated access token rather than the (now + // invalidated) one we just minted, so neither sync parks the connection. + const usableToken = await storeOuraTokens( + userId, + rotated.access_token, + rotated.refresh_token, + refreshTokenCiphertext, + ); + if (!usableToken) throw err; + return run(usableToken); } } @@ -202,6 +212,7 @@ export async function syncUserOura( userId, conn.accessToken, conn.refreshToken, + conn.refreshTokenCiphertext, opts.lookbackDays ?? OURA_SYNC_LOOKBACK_DAYS, ); } catch (err) { diff --git a/src/lib/tz/__tests__/local-day.test.ts b/src/lib/tz/__tests__/local-day.test.ts index f116dd3bb..e9eac4f8c 100644 --- a/src/lib/tz/__tests__/local-day.test.ts +++ b/src/lib/tz/__tests__/local-day.test.ts @@ -1,6 +1,46 @@ import { describe, expect, it } from "vitest"; import { startOfLocalDayInTz } from "@/lib/tz/local-day"; -import { wallClockInTz } from "@/lib/tz/wall-clock"; +import { wallClockInTz, zonedWallClockToUtc } from "@/lib/tz/wall-clock"; + +describe("zonedWallClockToUtc", () => { + it("resolves a wall clock in a positive-offset zone to the right UTC instant", () => { + // 00:30 CEST (UTC+2) on 2026-05-11 → 22:30 UTC on 2026-05-10. + const utc = zonedWallClockToUtc( + { year: 2026, month: 5, day: 11, hour: 0, minute: 30 }, + "Europe/Berlin", + ); + expect(utc.toISOString()).toBe("2026-05-10T22:30:00.000Z"); + }); + + it("round-trips through wallClockInTz across the spring-forward boundary", () => { + // 03:30 local on Berlin's spring-forward day (clocks jumped 02:00→03:00). + const utc = zonedWallClockToUtc( + { year: 2026, month: 3, day: 29, hour: 3, minute: 30 }, + "Europe/Berlin", + ); + const parts = wallClockInTz(utc, "Europe/Berlin"); + expect(parts.hour).toBe(3); + expect(parts.minute).toBe(30); + expect(parts.day).toBe(29); + }); + + it("resolves a negative-offset zone", () => { + // 00:30 NZST (UTC+12) on 2026-06-15 → 12:30 UTC on 2026-06-14. + const utc = zonedWallClockToUtc( + { year: 2026, month: 6, day: 15, hour: 0, minute: 30 }, + "Pacific/Auckland", + ); + expect(utc.toISOString()).toBe("2026-06-14T12:30:00.000Z"); + }); + + it("falls back to host-local when tz is undefined", () => { + const utc = zonedWallClockToUtc( + { year: 2026, month: 6, day: 15, hour: 14, minute: 30, second: 15 }, + undefined, + ); + expect(utc.getTime()).toBe(new Date(2026, 5, 15, 14, 30, 15).getTime()); + }); +}); describe("startOfLocalDayInTz", () => { it("anchors at the user's local midnight, not UTC midnight", () => { diff --git a/src/lib/tz/wall-clock.ts b/src/lib/tz/wall-clock.ts index 4dc5efcf5..68df50b26 100644 --- a/src/lib/tz/wall-clock.ts +++ b/src/lib/tz/wall-clock.ts @@ -64,6 +64,75 @@ const WALL_CLOCK_OPTIONS: Omit = weekday: "short", }); +/** + * The wall-clock fields a `zonedWallClockToUtc` caller supplies. Months are + * 1-based to match `WallClockParts`; second defaults to 0 when omitted. + */ +export interface WallClockInput { + year: number; + /** 1–12 (January = 1, December = 12). */ + month: number; + /** 1–31. */ + day: number; + /** 0–23. */ + hour: number; + minute: number; + second?: number; +} + +/** + * Inverse of `wallClockInTz`: given the wall-clock an observer in `tz` reads + * (e.g. an offset-less `2020-01-26T03:02:30` Fitbit sleep timestamp), return the + * UTC instant it denotes. DST-safe via the same two-pass converge as + * `startOfLocalDayInTz` — the second pass settles the zone offset across a DST + * transition for every IANA zone. When `tz` is omitted the parts are read as the + * host's system-local clock so offset-less callers keep their legacy shape. + */ +export function zonedWallClockToUtc( + parts: WallClockInput, + tz: string | undefined, +): Date { + const second = parts.second ?? 0; + if (!tz) { + return new Date( + parts.year, + parts.month - 1, + parts.day, + parts.hour, + parts.minute, + second, + 0, + ); + } + // Treat the wall clock as if it were UTC, then correct by the zone offset at + // that approximate instant; one extra pass settles the DST boundary. + const asIfUtc = Date.UTC( + parts.year, + parts.month - 1, + parts.day, + parts.hour, + parts.minute, + second, + 0, + ); + let guess = new Date(asIfUtc); + for (let i = 0; i < 2; i++) { + const at = wallClockInTz(guess, tz); + const guessAsIfUtc = Date.UTC( + at.year, + at.month - 1, + at.day, + at.hour, + at.minute, + at.second, + 0, + ); + const offsetMin = Math.round((guessAsIfUtc - guess.getTime()) / 60_000); + guess = new Date(asIfUtc - offsetMin * 60_000); + } + return guess; +} + export function wallClockInTz( date: Date, tz: string | undefined, diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index 33d5afd2f..bec3e8df8 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -20,6 +20,7 @@ const { whoopConnection: { findUnique: vi.fn(), update: vi.fn(), + updateMany: vi.fn(), findMany: vi.fn(), }, measurement: { @@ -118,7 +119,7 @@ describe("getValidToken — rotating refresh", () => { refresh_token: "new-refresh", expires_in: 3600, }); - prismaMock.whoopConnection.update.mockResolvedValue({}); + prismaMock.whoopConnection.updateMany.mockResolvedValue({ count: 1 }); const result = await getValidToken("user1"); @@ -127,11 +128,39 @@ describe("getValidToken — rotating refresh", () => { clientId: "cid", clientSecret: "csecret", }); - const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0]; + const updateArg = prismaMock.whoopConnection.updateMany.mock.calls[0]![0]; + // CAS guard: scoped to the connection AND the exact stored ciphertext spent. + expect(updateArg.where.id).toBe("conn1"); + expect(updateArg.where.refreshToken).toBe("enc(old-refresh)"); expect(updateArg.data.accessToken).toBe("enc(new-access)"); expect(updateArg.data.refreshToken).toBe("enc(new-refresh)"); }); + it("reuses the peer's rotated token on a lost CAS race (no spurious reauth)", async () => { + prismaMock.whoopConnection.findUnique + .mockResolvedValueOnce({ + id: "conn1", + whoopUserId: "42", + accessToken: "enc(old-access)", + refreshToken: "enc(old-refresh)", + tokenExpiresAt: new Date(Date.now() - 1000), + }) + // Re-read after the lost race surfaces the peer's freshly rotated access. + .mockResolvedValueOnce({ accessToken: "enc(peer-access)" }); + refreshAccessTokenMock.mockResolvedValue({ + access_token: "new-access", + refresh_token: "new-refresh", + expires_in: 3600, + }); + // A concurrent sync already rotated the token → zero rows match the guard. + prismaMock.whoopConnection.updateMany.mockResolvedValue({ count: 0 }); + + const result = await getValidToken("user1"); + + expect(result?.accessToken).toBe("peer-access"); + expect(recordSyncFailure).not.toHaveBeenCalled(); + }); + it("returns the stored token without refresh when not near expiry", async () => { prismaMock.whoopConnection.findUnique.mockResolvedValue({ id: "conn1", diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index baf009137..5015819ea 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -6,7 +6,9 @@ * - `getValidToken` decrypts the stored pair, refreshes at * `tokenExpiresAt - 5 min`, and persists BOTH rotated tokens (WHOOP * invalidates the prior access AND refresh token on every refresh — the - * same discipline Withings uses for its rotating refresh token). + * same discipline Withings uses for its rotating refresh token). The persist + * is a compare-and-swap on the stored ciphertext so two overlapping syncs + * cannot park the connection at reauth (see `persistRotatedToken`). * - Each per-resource sync (`sync-recovery` / `sync-sleep` / `sync-cycle` / * `sync-workout`) upserts into `Measurement` / `Workout` keyed on * `(userId, type, source = WHOOP, externalId)` so a re-post (WHOOP @@ -38,6 +40,7 @@ import { recomputeBucketsForMeasurement, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { persistRotatedToken } from "@/lib/integrations/oauth-refresh"; import { refreshAccessToken } from "./client"; import { getUserWhoopCredentials } from "./credentials"; import { @@ -217,20 +220,44 @@ export async function getValidToken( const newTokens = await refreshAccessToken(refreshToken, creds); const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000); - // WHOOP rotates the refresh token on every refresh — persist BOTH the - // new access token AND the new refresh token, or the next refresh - // reuses an invalidated token and the connection drops to reauth. - await prisma.whoopConnection.update({ - where: { id: connection.id }, - data: { - accessToken: encrypt(newTokens.access_token), - refreshToken: encrypt(newTokens.refresh_token), - tokenExpiresAt: expiresAt, + // WHOOP rotates the refresh token on every refresh. Persist BOTH the new + // access token AND the new refresh token with a compare-and-swap on the + // stored ciphertext: a concurrent sync that already rotated the token is + // not clobbered, and THIS caller — if it lost the race — reuses the peer's + // rotated access token instead of dropping the connection to reauth. + const persistedAccessToken = await persistRotatedToken( + newTokens.access_token, + { + conditionalUpdate: async () => { + const { count } = await prisma.whoopConnection.updateMany({ + // `connection.refreshToken` is the exact ciphertext we read + + // spent; matching it is the CAS guard (a re-encrypt would differ). + where: { + id: connection.id, + refreshToken: connection.refreshToken, + }, + data: { + accessToken: encrypt(newTokens.access_token), + refreshToken: encrypt(newTokens.refresh_token), + tokenExpiresAt: expiresAt, + }, + }); + return count; + }, + readPeerAccessToken: async () => { + const fresh = await prisma.whoopConnection.findUnique({ + where: { id: connection.id }, + select: { accessToken: true }, + }); + return fresh ? decrypt(fresh.accessToken) : null; + }, }, - }); + ); + + if (!persistedAccessToken) return null; return { - accessToken: newTokens.access_token, + accessToken: persistedAccessToken, connection: { id: connection.id, whoopUserId: connection.whoopUserId,