/** * v1.17.0 (F4) — Polar AccessLink sync. * * Pulls Nightly Recharge (→ RECOVERY_SCORE / ANS_CHARGE / HRV / RHR / * respiratory rate), sleep (per-stage durations on a reconstructed END-instant * timeline + sleep score), daily activity (steps + active energy + distance), * Training Load Pro (→ CARDIO_LOAD), and SpO2 (→ OXYGEN_SATURATION) for one * connected user, mapping each into `Measurement` rows tagged `source = POLAR`. * * Token model: Polar tokens do not expire and have no refresh path, so there is * no `getValidToken` refresh dance (the WHOOP shape). A revoked grant surfaces * as a 401 → `reauth_required` on the shared integration ledger (`polar` key), * exactly like Nightscout's token-rejected path. * * Idempotency: each row's `externalId` is `:` and the write is * an upsert keyed on `(userId, type, source = POLAR, externalId)`. Polar * re-scores a night for a short window after the fact, so the `update` branch * overwrites in place (WHOOP-style re-score), not first-write-wins. * * The measurement-write tail (per-row upsert → rollup fold → status-insight * invalidate) mirrors the shared WHOOP / Nightscout sync tail; it is NOT a new * write path. */ import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; import { getEvent } from "@/lib/logging/context"; import { recordSyncFailure, recordSyncSuccess, toFailureKind, type FailureKind, } from "@/lib/integrations/status"; import { collapseToTypeDayKeys, recomputeBucketsForMeasurement, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { fetchActivities, fetchCardioLoads, fetchNightlyRecharges, fetchSleeps, fetchSpo2, mapActivity, mapCardioLoad, mapNightlyRecharge, mapSleep, mapSpo2, type MappedMeasurement, } from "./client"; import { getPolarConnection } from "./credentials"; import { PolarApiError, classifyPolarError } from "./response-classifier"; /** Map a Polar error onto the shared integration-ledger failure kind. */ export function classifyPolarFailure(err: unknown): FailureKind { return toFailureKind(classifyPolarError(err)); } /** One mapped reading with its `externalId` resolved. */ export interface PolarMeasurementUpsert { type: string; value: number; unit: string; measuredAt: Date; externalId: string; sleepStage?: "CORE" | "DEEP" | "REM" | "AWAKE" | "IN_BED" | null; } function toUpsert( mapped: MappedMeasurement[], resourcePrefix: string, ): PolarMeasurementUpsert[] { return mapped.map((m) => ({ type: m.type, value: m.value, unit: m.unit, measuredAt: m.measuredAt, // 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. externalId: m.externalId ?? `${resourcePrefix}:${m.measuredAt.toISOString().slice(0, 10)}:${m.fieldTag}`, sleepStage: m.sleepStage ?? null, })); } /** * Sync one user's Polar data. Returns the count of measurement rows written. * A user with no Polar connection is a clean no-op (returns 0, touches no * status row). */ export async function syncUserPolar(userId: string): Promise { const conn = await getPolarConnection(userId); if (!conn) return 0; const readings: PolarMeasurementUpsert[] = []; try { const [recharges, sleeps, activities, cardioLoads, spo2Tests] = await Promise.all([ fetchNightlyRecharges(conn.accessToken, conn.polarUserId), fetchSleeps(conn.accessToken, conn.polarUserId), fetchActivities(conn.accessToken, conn.polarUserId), fetchCardioLoads(conn.accessToken, conn.polarUserId), fetchSpo2(conn.accessToken, conn.polarUserId), ]); for (const r of recharges) { readings.push(...toUpsert(mapNightlyRecharge(r), "recharge")); } for (const s of sleeps) { readings.push(...toUpsert(mapSleep(s), "sleep")); } for (const a of activities) { readings.push(...toUpsert(mapActivity(a), "activity")); } for (const c of cardioLoads) { readings.push(...toUpsert(mapCardioLoad(c), "cardioload")); } for (const t of spo2Tests) { readings.push(...toUpsert(mapSpo2(t), "spo2")); } } catch (err) { await recordSyncFailure({ userId, integration: "polar", kind: classifyPolarFailure(err), message: err instanceof Error ? err.message : String(err), errorCode: err instanceof PolarApiError && err.httpStatus != null ? String(err.httpStatus) : undefined, }); throw err; } const imported = await upsertPolarMeasurements(userId, readings); await recordSyncSuccess(userId, "polar"); return imported; } /** * Upsert a batch of mapped Polar readings, then fold the rollup tier + * invalidate status-insight caches once at the end (mirrors the WHOOP / * Nightscout sync tail). Idempotent: the `(userId, type, source, externalId)` * unique key makes a re-post overwrite in place. Best-effort on the rollup fold. */ export async function upsertPolarMeasurements( userId: string, readings: PolarMeasurementUpsert[], ): Promise { if (readings.length === 0) return 0; let imported = 0; const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; for (const r of readings) { const type = r.type as MeasurementType; try { await prisma.measurement.upsert({ where: { userId_type_source_externalId: { userId, type, source: "POLAR", externalId: r.externalId, }, }, create: { userId, type, source: "POLAR", value: r.value, unit: r.unit, measuredAt: r.measuredAt, externalId: r.externalId, sleepStage: r.sleepStage ?? null, }, update: { value: r.value, unit: r.unit, measuredAt: r.measuredAt, sleepStage: r.sleepStage ?? null, syncVersion: { increment: 1 }, }, }); touched.push({ type, measuredAt: r.measuredAt }); imported++; } catch (err) { getEvent()?.addWarning(`polar: failed to upsert measurement: ${err}`); } } try { const keys = collapseToTypeDayKeys(touched); for (const k of keys) { await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); } invalidateStatusInsightsForTypes( userId, keys.map((k) => k.type), ).catch((err) => { getEvent()?.addWarning( `polar: status-insight invalidate failed for ${userId}: ${err}`, ); }); } catch (err) { getEvent()?.addWarning( `polar: rollup recompute failed for ${userId}: ${err}`, ); } return imported; }