Files
HealthLog/src/lib/fitbit/sync-activity.ts
T
Marc-André Bombeck 3021d54068 fix(fitbit): skip tombstones on sync, batch backfill writes, bound poll concurrency
Three hardening + performance fixes to the Fitbit/Google Health sync.

Stop soft-delete resurrection. The measurement write probed the existing
row by the `(userId, type, source, externalId)` unique key without excluding
soft-deleted rows, so a reading the user deleted came back on the next hourly
sync (the upsert matched the tombstone and took the update branch — silently
undoing the delete). The write now probes only live rows (`deletedAt: null`)
and treats a tombstone as absent, minting a fresh insert instead. The fresh
insert relies on the partial unique index over `deleted_at IS NULL` to keep
live-row uniqueness; the tombstone sits outside it, and `skipDuplicates` guards
a live-row race.

Batch the backfill writes + collapse the rollup recompute. The write loop
was one upsert per reading (N+1) followed by a per-(type,day) DAY-rollup
recompute, so a multi-year backfill paid thousands of serial round-trips on a
concurrency-1 worker. The existence probe is now a single `findMany` over the
batch; fresh rows go through chunked `createMany`; only already-live rows take a
per-row update for their differing values. On a `fullSync` the per-write rollup
hook is deferred and the touched type-days collapse into one
`recomputeUserRollups(from, to)` pass at the end of the cycle. The incremental
path keeps the inline per-day hook. The `stats:` overwrite contract, the
single-watermark `markSynced`, and the all-403 soft-skip guard are unchanged.

Bound the hourly poll concurrency. The poll synced every connection in one
serial loop, so a single slow Google response stalled the whole cohort. The
cohort now fans out through a bounded `p-limit` pool with per-user error
isolation, extracted into `runFitbitPollCohort` so the contract is unit testable
without exporting worker internals. The accumulators are read after the await to
avoid a compound-assignment lost-update across overlapping pool tasks.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
(cherry picked from commit 8369af2471fb8da04a938b5a784433318d81a248)
2026-06-05 03:01:36 +02:00

127 lines
4.2 KiB
TypeScript

/**
* Fitbit / Google Health activity-bundle sync (v1.12.0, W5).
*
* Reads the daily cumulative-activity data types from the
* `activity_and_fitness.readonly` Restricted bundle and upserts each mapped
* daily total as `source = FITBIT`:
*
* - steps → ACTIVITY_STEPS (count)
* - distance → WALKING_RUNNING_DISTANCE (metres)
* - active-calories → ACTIVE_ENERGY_BURNED (kcal — ACTIVE portion only)
* - floors → FLIGHTS_CLIMBED (count)
* - vo2-max → VO2_MAX (mL/(kg·min); daily latest-wins)
*
* These are per-day summaries (one value per calendar day). The externalId is
* minted with the `stats:` daily-total prefix — `stats:<fieldTag>:<YYYY-MM-DD>`
* — so a re-fetched day OVERWRITES the existing row rather than minting a
* duplicate, matching the Apple-Health `stats:<HK>:<YYYY-MM-DD>` overwrite
* contract. A day of rest legitimately records 0 steps / 0 floors / 0 active
* kcal, so the cumulative mappers preserve a zero (dropping it would leave a
* hole the chart misreads as missing data); VO2 max stays strictly positive.
*
* A per-data-class 403 soft-skips THAT class (returns 0, leaves the connection
* connected) — the six Restricted bundles are granted independently.
*/
import {
FITBIT_DATA_TYPES,
type FitbitDataType,
type FitbitMappedMeasurement,
fetchDataPoints,
mapActiveCalories,
mapDistance,
mapFloors,
mapSteps,
mapVo2Max,
} from "./client";
import {
getValidToken,
handleCollectionFetchError,
upsertFitbitMeasurements,
type FitbitMeasurementUpsert,
type FitbitResourceSyncOptions,
} from "./sync";
import { annotate } from "@/lib/logging/context";
/** One mappable activity metric: its data-type encoding + the per-point mapper + a verb. */
interface ActivityResource {
dataType: FitbitDataType;
map: (point: Record<string, unknown>) => FitbitMappedMeasurement[];
verb: string;
}
const ACTIVITY_RESOURCES: ActivityResource[] = [
{ dataType: FITBIT_DATA_TYPES.steps, map: mapSteps, verb: "fetchSteps" },
{
dataType: FITBIT_DATA_TYPES.distance,
map: mapDistance,
verb: "fetchDistance",
},
{
dataType: FITBIT_DATA_TYPES.activeCalories,
map: mapActiveCalories,
verb: "fetchActiveCalories",
},
{ dataType: FITBIT_DATA_TYPES.floors, map: mapFloors, verb: "fetchFloors" },
{ dataType: FITBIT_DATA_TYPES.vo2Max, map: mapVo2Max, verb: "fetchVo2Max" },
];
/**
* Assemble the stored externalId for a mapped activity reading. A cumulative
* daily metric carries the `cumulativeDaily` flag from its mapper — its
* externalId gets the `stats:` daily-total prefix so a re-fetched day overwrites
* in place (the mapper's `fieldTag` is already the `<tag>:<YYYY-MM-DD>` form).
*/
function externalIdFor(m: FitbitMappedMeasurement): string {
return m.cumulativeDaily ? `stats:${m.fieldTag}` : m.fieldTag;
}
export async function syncUserActivity(
userId: string,
opts: FitbitResourceSyncOptions = {},
): Promise<number> {
const tokenInfo = await getValidToken(userId);
if (!tokenInfo) return 0;
// Cycle-wide watermark snapshotted once by `syncUserFitbit`; undefined on a
// full/backfill run.
const start = opts.start;
let imported = 0;
for (const resource of ACTIVITY_RESOURCES) {
let points: Record<string, unknown>[];
try {
points = await fetchDataPoints(
resource.dataType,
tokenInfo.accessToken,
resource.verb,
{ start },
);
} catch (err) {
imported += await handleCollectionFetchError(resource.verb, userId, err);
continue;
}
const readings: FitbitMeasurementUpsert[] = [];
for (const point of points) {
for (const m of resource.map(point)) {
readings.push({
type: m.type,
value: m.value,
unit: m.unit,
measuredAt: m.measuredAt,
externalId: externalIdFor(m),
});
}
}
imported += (
await upsertFitbitMeasurements(userId, readings, {
deferRollup: opts.deferRollup,
})
).imported;
}
// `markSynced` is owned by the orchestrator (`syncUserFitbit`).
annotate({ action: { name: "fitbit.activity.sync", details: { imported } } });
return imported;
}