feat(measurements): take hourly sums of summed types into the intraday profile and a backfill trigger that moves no live stamp; fix(backup): restore qualitative lab markers

This commit is contained in:
Marc-André Bombeck committed 2026-10-16 00:18:35 +02:00
1 parent 79f7eb199b
commit a0f6b89e92
34 files changed
+2468 -286

No files matched your search

+2 -2
View File
@@ -87,8 +87,8 @@ const SERVER_FEATURES = {
trackIntake: true,
// v1.43 — each turns true in the change that makes its route or request
// form work; until then a client must not send it (contract §0.6).
cumulativeHourBuckets: false,
syncTriggerBackfill: false,
cumulativeHourBuckets: true,
syncTriggerBackfill: true,
biomarkerMerge: true,
lifeCircumstances: false,
} as const;
@@ -533,10 +533,72 @@ describe("GET /api/integrations/healthkit — sync health", () => {
lastTrigger: "foreground",
lastNewSampleAt: null,
lastBackgroundAt: backgroundAt.toISOString(),
lastBackfillAt: null,
});
expect(byType.PULSE).toMatchObject({ lastBackgroundAt: null });
});
// v1.43 (ios #77) — a backfill's own ledger row reads as `lastBackfillAt`
// of its type and moves none of the live facts.
it("serves lastBackfillAt from the backfill row, apart from the live arrival", async () => {
vi.mocked(prisma.user.findUnique).mockResolvedValue({
healthKitConfigJson: null,
healthKitLastSyncedAt: ago(60 * 60 * 1000),
} as never);
vi.mocked(listLiveMeasurementTypes).mockResolvedValue([
"PULSE",
"ACTIVITY_STEPS",
]);
mockProbes([
{ type: "PULSE", last_seen: ago(60 * 60 * 1000) },
{ type: "ACTIVITY_STEPS", last_seen: ago(60 * 60 * 1000) },
]);
const receivedAt = ago(5 * 60 * 60 * 1000);
const backfillAt = ago(60 * 1000);
vi.mocked(prisma.healthKitTypeSync.findMany).mockResolvedValue([
{
type: "PULSE",
lastReceivedAt: receivedAt,
lastTrigger: "foreground",
lastNewSampleAt: receivedAt,
lastBackgroundAt: null,
},
{
type: "backfill:PULSE",
lastReceivedAt: backfillAt,
lastTrigger: null,
lastNewSampleAt: null,
lastBackgroundAt: null,
},
{
type: "backfill:ACTIVITY_STEPS",
lastReceivedAt: backfillAt,
lastTrigger: null,
lastNewSampleAt: null,
lastBackgroundAt: null,
},
] as never);
const body = await read();
const byType = Object.fromEntries(
body.metricFreshness.map((entry) => [entry.type, entry]),
);
expect(byType.PULSE).toMatchObject({
lastReceivedAt: receivedAt.toISOString(),
lastTrigger: "foreground",
lastBackfillAt: backfillAt.toISOString(),
});
// A type that only ever arrived through a backfill has no live arrival.
expect(byType.ACTIVITY_STEPS).toMatchObject({
lastReceivedAt: null,
lastTrigger: null,
lastBackfillAt: backfillAt.toISOString(),
});
expect(body.metricFreshness.map((e) => e.type)).not.toContain(
"backfill:PULSE",
);
});
it("leaves the dose entry out when there is no Apple Health dose", async () => {
vi.mocked(prisma.user.findUnique).mockResolvedValue({
healthKitConfigJson: null,
+200 -26
View File
@@ -73,10 +73,19 @@ import {
findFoldedWindowDuplicates,
} from "@/lib/measurements/folded-window";
import {
HEALTHKIT_SYNC_TRIGGERS,
BACKFILL_SYNC_TRIGGER,
HEALTHKIT_REQUEST_SYNC_TRIGGERS,
recordHealthKitTypeArrivals,
type HealthKitTypeArrival,
} from "@/lib/integrations/healthkit-type-sync";
import {
classifyHourBucket,
HOUR_BUCKET_SKIP_REASONS,
parseHourBucketId,
writeHourBuckets,
type HourBucketWrite,
} from "@/lib/measurements/intraday-hour-buckets";
import { resolveStoredTimezone } from "@/lib/tz/resolver";
import { isPlausibleEntryInstant } from "@/lib/validations/entry-instant";
import {
classifyExternalId,
@@ -199,7 +208,9 @@ const batchEntrySchema = z.object({
// pre-#66 caller stays byte-for-byte unchanged.
// v1.42 (#1173) — `manual` names a user-initiated "Sync all" run, as opposed to
// the app simply being open (`foreground`).
const syncTriggerEnum = z.enum(HEALTHKIT_SYNC_TRIGGERS);
// v1.43 (ios #77) — `backfill` names a history catch-up. It moves only the
// per-type `lastBackfillAt`, never the live sync stamps (see below).
const syncTriggerEnum = z.enum(HEALTHKIT_REQUEST_SYNC_TRIGGERS);
const batchPayloadSchema = z.object({
entries: z.array(batchEntrySchema).min(1),
@@ -380,6 +391,11 @@ async function postBatch(request: NextRequest): Promise<Response> {
const prepared: Prepared[] = [];
const results: EntryResult[] = new Array(entries.length);
const unstableShapes: UnstableExternalIdShape[] = [];
// v1.43 — hourly sums for the intraday profile, and the zone they are cut in
// (the account's, the one `/api/auth/me` publishes as `timezone`).
const hourWrites: HourBucketWrite[] = [];
let accountTz: string | undefined;
const batchNow = new Date();
for (let index = 0; index < entries.length; index++) {
const entry = entries[index];
@@ -402,6 +418,75 @@ async function postBatch(request: NextRequest): Promise<Response> {
continue;
}
// v1.43 (ios #77) — an hourly sum of a summed type
// (`stats:<HK>:<YYYY-MM-DD>T<HH>`) is never a measurement: it lands in the
// intraday profile behind the same-time baseline, and the day's total
// stays the day row. Validated here, written after the measurements.
const hourId = parseHourBucketId(entry.externalId);
if (hourId) {
const skip = (reason: string) => {
results[index] = { index, status: "skipped", reason };
};
// The profile has no source dimension and is the phone's: a bridge or a
// hand-entered row cannot post hours into it.
if (scoped || (entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") {
skip(HOUR_BUCKET_SKIP_REASONS.unsupportedType);
continue;
}
accountTz ??= await resolveStoredTimezone(user.timezone);
const startDate = new Date(entry.startDate);
const verdict = classifyHourBucket(
hourId,
startDate,
accountTz,
batchNow,
);
if (!verdict.ok) {
skip(verdict.reason);
continue;
}
if (!isPlausibleEntryInstant(startDate)) {
skip("implausible_timestamp");
continue;
}
const hourMapped = mapAppleHealthEntry({
hkIdentifier: entry.hkIdentifier,
value: entry.value,
unit: entry.unit,
startDate: entry.startDate,
endDate: entry.endDate,
});
if (!hourMapped || hourMapped.type !== verdict.type) {
// The entry's identifier must be the one its id names.
skip(
hourMapped
? HOUR_BUCKET_SKIP_REASONS.unsupportedType
: "unmappable_identifier",
);
continue;
}
if (
hourMapped.value < 0 ||
validateMeasurementRange(hourMapped.type, hourMapped.value) !== null
) {
results[index] = outOfRangeResult(
index,
hourMapped.type,
hourMapped.value,
);
continue;
}
hourWrites.push({
index,
type: verdict.type,
hkIdentifier: hourId.hkIdentifier,
dateKey: hourId.dateKey,
hour: hourId.hour,
value: hourMapped.value,
});
continue;
}
const mapped = mapAppleHealthEntry({
hkIdentifier: entry.hkIdentifier,
value: entry.value,
@@ -898,15 +983,79 @@ async function postBatch(request: NextRequest): Promise<Response> {
}
}
// v1.43 — the hours, after the measurements. The last entry for one slot
// wins, as for the day rows; an earlier one in the same batch is a
// superseded duplicate.
let hourInsertedCount = 0;
let hourUpdatedCount = 0;
const hourLanded: HourBucketWrite[] = [];
if (hourWrites.length > 0 && accountTz) {
const lastBySlot = new Map<string, HourBucketWrite>();
for (const w of hourWrites) {
lastBySlot.set(`${w.type}::${w.dateKey}::${w.hour}`, w);
}
for (const w of hourWrites) {
if (lastBySlot.get(`${w.type}::${w.dateKey}::${w.hour}`) !== w) {
results[w.index] = {
index: w.index,
status: "duplicate",
reason: "superseded_in_batch",
};
duplicateCount++;
}
}
const toWrite = [...lastBySlot.values()];
try {
const statuses = await writeHourBuckets(
prisma,
user.id,
accountTz,
toWrite,
);
for (const w of toWrite) {
const status = statuses.get(w.index) ?? "inserted";
results[w.index] = { index: w.index, status };
if (status === "inserted") hourInsertedCount++;
else hourUpdatedCount++;
hourLanded.push(w);
}
} catch (err) {
for (const w of toWrite) {
results[w.index] = {
index: w.index,
status: "failed",
reason: "write_failed",
};
failedCount++;
}
annotate({
action: { name: "measurement.batch.hour-bucket.failed" },
meta: { error: err instanceof Error ? err.message : String(err) },
});
}
annotate({
action: { name: "measurement.batch.hour-bucket" },
meta: {
attempted: hourWrites.length,
inserted: hourInsertedCount,
updated: hourUpdatedCount,
processed: entries.length,
},
});
}
const healthKitSyncSucceeded =
failedCount === 0 &&
prepared.some((p) => {
if ((p.entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") return false;
const status = results[p.index]?.status;
return (
status === "inserted" || status === "updated" || status === "duplicate"
);
});
(hourLanded.length > 0 ||
prepared.some((p) => {
if ((p.entry.source ?? "APPLE_HEALTH") !== "APPLE_HEALTH") return false;
const status = results[p.index]?.status;
return (
status === "inserted" ||
status === "updated" ||
status === "duplicate"
);
}));
// This batch route is the native ingestion boundary, so it is also where the
// SyncMode checkpoint (`User.lastSyncedAt`, the same column
@@ -938,6 +1087,11 @@ async function postBatch(request: NextRequest): Promise<Response> {
// uses to size its undelivered window, so letting a scale advance it can make
// the phone skip a window of its own samples — a silent data loss whose cause
// would be nowhere near the code that caused it.
//
// v1.43 (ios #77) — a `backfill` batch is a catch-up, not a sync. It records
// its types under `lastBackfillAt` only and moves none of the live stamps:
// thirty days of history must not read as thirty healthy daily syncs.
const backfill = syncTrigger === BACKFILL_SYNC_TRIGGER;
if (failedCount === 0 && !scoped) {
const syncedAt = new Date();
const deliveredWithoutTheApp =
@@ -962,6 +1116,13 @@ async function postBatch(request: NextRequest): Promise<Response> {
if (status !== "duplicate") entry.newSamples += 1;
arrivals.set(type, entry);
}
// The hours the phone posted are an arrival of their type as well.
for (const w of hourLanded) {
const entry = arrivals.get(w.type) ?? { accepted: 0, newSamples: 0 };
entry.accepted += 1;
entry.newSamples += 1;
arrivals.set(w.type, entry);
}
if (arrivals.size > 0) {
await recordHealthKitTypeArrivals(prisma, {
userId: user.id,
@@ -970,21 +1131,23 @@ async function postBatch(request: NextRequest): Promise<Response> {
at: syncedAt,
});
}
await prisma.user.update({
where: { id: user.id },
data: {
lastSyncedAt: syncedAt,
...(healthKitSyncSucceeded
? {
healthKitLastSyncedAt: syncedAt,
healthKitLastSyncTrigger: syncTrigger ?? null,
...(deliveredWithoutTheApp
? { healthKitLastBackgroundSyncAt: syncedAt }
: {}),
}
: {}),
},
});
if (!backfill) {
await prisma.user.update({
where: { id: user.id },
data: {
lastSyncedAt: syncedAt,
...(healthKitSyncSucceeded
? {
healthKitLastSyncedAt: syncedAt,
healthKitLastSyncTrigger: syncTrigger ?? null,
...(deliveredWithoutTheApp
? { healthKitLastBackgroundSyncAt: syncedAt }
: {}),
}
: {}),
},
});
}
}
const freshlyInsertedRows = insertedPrepared.map((p) => ({
@@ -1251,10 +1414,21 @@ async function postBatch(request: NextRequest): Promise<Response> {
}
}
// v1.43 — the hours are not measurements, so they touched none of the
// measurement consumers above; only the caches that read the intraday
// profile (the same-time baseline) are marked stale.
if (
hourInsertedCount + hourUpdatedCount > 0 &&
insertedCount === 0 &&
updatedCount === 0
) {
invalidateUserMeasurements(user.id);
}
const response = apiSuccess({
processed: entries.length,
inserted: insertedCount,
updated: updatedCount,
inserted: insertedCount + hourInsertedCount,
updated: updatedCount + hourUpdatedCount,
duplicates: duplicateCount,
failed: failedCount,
skipped,
@@ -80,7 +80,9 @@ const USER = {
};
/** A replayed Apple dose: the cheapest batch that still counts as an arrival. */
async function postReplayedDose(): Promise<Response> {
async function postReplayedDose(
syncTrigger: string = "background",
): Promise<Response> {
vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValueOnce([
{ id: "row-existing", externalId: "hk-dose-1" },
] as never);
@@ -89,7 +91,7 @@ async function postReplayedDose(): Promise<Response> {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
syncTrigger: "background",
syncTrigger,
entries: [
{
medicationId: "med-apple",
@@ -134,4 +136,33 @@ describe("the dose arrival record and the credential that sent the batch", () =>
expect(res.status).toBe(200);
expect(recordHealthKitTypeArrivals).not.toHaveBeenCalled();
});
// v1.43 (ios #77) — a history catch-up records the dose arrival under
// `lastBackfillAt` only, and does not move the sync checkpoint.
it("passes a backfill through and leaves the checkpoint where it was", async () => {
vi.mocked(requireAuth).mockResolvedValue({
session: { id: "s" },
user: USER,
} as never);
const res = await postReplayedDose("backfill");
expect(res.status).toBe(200);
expect(recordHealthKitTypeArrivals).toHaveBeenCalledWith(
expect.anything(),
expect.objectContaining({ trigger: "backfill" }),
);
expect(prisma.user.update).not.toHaveBeenCalled();
});
it("moves the checkpoint for a live sync", async () => {
vi.mocked(requireAuth).mockResolvedValue({
session: { id: "s" },
user: USER,
} as never);
await postReplayedDose("background");
expect(prisma.user.update).toHaveBeenCalledWith(
expect.objectContaining({
data: expect.objectContaining({ lastSyncedAt: expect.any(Date) }),
}),
);
});
});
+9 -4
View File
@@ -58,7 +58,8 @@ import { withIdempotency } from "@/lib/idempotency";
import { checkRateLimit } from "@/lib/rate-limit";
import { markSyncCheckpoint } from "@/lib/sync/checkpoint";
import {
HEALTHKIT_SYNC_TRIGGERS,
BACKFILL_SYNC_TRIGGER,
HEALTHKIT_REQUEST_SYNC_TRIGGERS,
MEDICATION_DOSE_ARRIVAL_TYPE,
recordHealthKitTypeArrivals,
} from "@/lib/integrations/healthkit-type-sync";
@@ -176,7 +177,9 @@ const bulkPayloadSchema = z.object({
// v1.42.4 — what woke the phone for this sync, the same diagnostic-only
// field `POST /api/measurements/batch` takes. It changes no row; it only
// feeds the `MEDICATION_DOSE` arrival record below.
syncTrigger: z.enum(HEALTHKIT_SYNC_TRIGGERS).optional(),
// v1.43 (ios #77) — `backfill` for a history catch-up: it moves only the
// dose entry's `lastBackfillAt`, never the live stamps or the checkpoint.
syncTrigger: z.enum(HEALTHKIT_REQUEST_SYNC_TRIGGERS).optional(),
});
// v1.8.2 — `updated` joins the per-entry status vocabulary: a write that
@@ -1038,8 +1041,10 @@ async function postBulk(request: NextRequest): Promise<Response> {
// The SyncMode checkpoint belongs to the push that moved the data, not to
// the `/api/sync/state` handshake that reports it. This endpoint is one of
// the three domains that handshake summarises, so an intake drain advances
// it.
await markSyncCheckpoint(user.id);
// it. A `backfill` catch-up does not: it is history, not a sync.
if (syncTrigger !== BACKFILL_SYNC_TRIGGER) {
await markSyncCheckpoint(user.id);
}
return apiSuccess({
processed: entries.length,