diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 1a61187f6..a44e05c43 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -81,8 +81,10 @@ import { import { classifyHourBucket, HOUR_BUCKET_SKIP_REASONS, + combineHourWrites, parseHourBucketId, writeHourBuckets, + type HourBucketEntry, type HourBucketWrite, } from "@/lib/measurements/intraday-hour-buckets"; import { resolveStoredTimezone } from "@/lib/tz/resolver"; @@ -393,7 +395,7 @@ async function postBatch(request: NextRequest): Promise { 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[] = []; + const hourWrites: HourBucketEntry[] = []; let accountTz: string | undefined; const batchNow = new Date(); @@ -483,6 +485,7 @@ async function postBatch(request: NextRequest): Promise { dateKey: hourId.dateKey, hour: hourId.hour, value: hourMapped.value, + startMs: startDate.getTime(), }); continue; } @@ -983,28 +986,24 @@ async function postBatch(request: NextRequest): Promise { } } - // 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. + // v1.43 — the hours, after the measurements. One write per slot: the last + // entry for one start wins (an earlier one in the same batch is a + // superseded duplicate), and the two starts of the hour a fall-back day + // repeats are summed (`combineHourWrites`). let hourInsertedCount = 0; let hourUpdatedCount = 0; const hourLanded: HourBucketWrite[] = []; if (hourWrites.length > 0 && accountTz) { - const lastBySlot = new Map(); - for (const w of hourWrites) { - lastBySlot.set(`${w.type}::${w.dateKey}::${w.hour}`, w); + const { writes, superseded } = combineHourWrites(hourWrites); + for (const index of superseded) { + results[index] = { + index, + status: "duplicate", + reason: "superseded_in_batch", + }; + duplicateCount++; } - 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()]; + const toWrite = writes.map(({ write }) => write); try { const statuses = await writeHourBuckets( prisma, @@ -1012,21 +1011,25 @@ async function postBatch(request: NextRequest): Promise { 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); + for (const { write, contributors } of writes) { + const status = statuses.get(write.index) ?? "inserted"; + for (const index of contributors) { + results[index] = { index, status }; + if (status === "inserted") hourInsertedCount++; + else hourUpdatedCount++; + } + hourLanded.push(write); } } catch (err) { - for (const w of toWrite) { - results[w.index] = { - index: w.index, - status: "failed", - reason: "write_failed", - }; - failedCount++; + for (const { contributors } of writes) { + for (const index of contributors) { + results[index] = { + index, + status: "failed", + reason: "write_failed", + }; + failedCount++; + } } annotate({ action: { name: "measurement.batch.hour-bucket.failed" }, diff --git a/src/lib/measurements/__tests__/intraday-hour-buckets.test.ts b/src/lib/measurements/__tests__/intraday-hour-buckets.test.ts index e7008f973..53eda2b47 100644 --- a/src/lib/measurements/__tests__/intraday-hour-buckets.test.ts +++ b/src/lib/measurements/__tests__/intraday-hour-buckets.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { classifyHourBucket, + combineHourWrites, cumulativeFromIncrements, HOUR_NOT_POSTED, isStartOfLocalHour, @@ -158,3 +159,38 @@ describe("cumulativeFromIncrements", () => { expect(postedHours(increments)).toBe(3); }); }); + +describe("combineHourWrites", () => { + const entry = (index: number, startIso: string, value: number) => ({ + index, + type: "ACTIVITY_STEPS" as const, + hkIdentifier: "HKQuantityTypeIdentifierStepCount", + dateKey: "2026-10-25", + hour: 2, + value, + startMs: Date.parse(startIso), + }); + + it("sums the two starts of the hour a fall-back day repeats", () => { + // Berlin, 2026-10-25: 02:00 CEST is 00:00Z, 02:00 CET is 01:00Z. + const { writes, superseded } = combineHourWrites([ + entry(0, "2026-10-25T00:00:00Z", 300), + entry(1, "2026-10-25T01:00:00Z", 120), + ]); + expect(superseded).toEqual([]); + expect(writes).toHaveLength(1); + expect(writes[0].write.value).toBe(420); + expect(writes[0].contributors.sort()).toEqual([0, 1]); + }); + + it("lets the last entry of one start win, and marks the earlier one", () => { + const { writes, superseded } = combineHourWrites([ + entry(0, "2026-10-25T00:00:00Z", 300), + entry(1, "2026-10-25T00:00:00Z", 310), + entry(2, "2026-10-25T01:00:00Z", 120), + ]); + expect(superseded).toEqual([0]); + expect(writes[0].write.value).toBe(430); + expect(writes[0].write.index).toBe(2); + }); +}); diff --git a/src/lib/measurements/intraday-hour-buckets.ts b/src/lib/measurements/intraday-hour-buckets.ts index 0c9940b29..2bd5a0830 100644 --- a/src/lib/measurements/intraday-hour-buckets.ts +++ b/src/lib/measurements/intraday-hour-buckets.ts @@ -177,6 +177,59 @@ export interface HourBucketWrite { value: number; } +/** One hour as the batch reads it, with the instant it starts at. */ +export interface HourBucketEntry extends HourBucketWrite { + /** `startDate` of the entry, in epoch milliseconds. */ + startMs: number; +} + +/** + * A batch's hours reduced to one write per slot. + * + * A slot is one local hour of one day. On the autumn day the clock falls + * back, one local hour happens twice, and the phone hands over both under + * the same key (`…T02`), each with its own `startDate`: they are two hours of + * activity and are summed into the slot rather than the second replacing the + * first. Entries that repeat the same start are the same hour posted twice, + * and the last one counts, as for every other slot. + * + * `contributors` are the entries whose values the write carries (they take + * its status); `superseded` are the earlier repeats of one start. + * + * The slot holds one number, so the sum is only as complete as the batch + * that carries it: both occurrences have to arrive together. The phone posts + * a day's hours in one batch, and a later batch that carries only one of + * them replaces the slot with that one, as a re-post of any slot does. + */ +export function combineHourWrites(entries: readonly HourBucketEntry[]): { + writes: Array<{ write: HourBucketWrite; contributors: number[] }>; + superseded: number[]; +} { + const slots = new Map>(); + const superseded: number[] = []; + for (const entry of entries) { + const key = `${entry.type}::${entry.dateKey}::${entry.hour}`; + const byStart = slots.get(key) ?? new Map(); + const earlier = byStart.get(entry.startMs); + if (earlier) superseded.push(earlier.index); + byStart.set(entry.startMs, entry); + slots.set(key, byStart); + } + const writes = [...slots.values()].map((byStart) => { + const occurrences = [...byStart.values()]; + const last = occurrences.reduce((a, b) => (b.index > a.index ? b : a)); + const { startMs: _start, ...base } = last; + return { + write: { + ...base, + value: occurrences.reduce((sum, entry) => sum + entry.value, 0), + }, + contributors: occurrences.map((entry) => entry.index), + }; + }); + return { writes, superseded }; +} + /** * Store a batch's hours, one transaction per (type, day). Returns the * per-entry status: `inserted` for the first value of a slot, `updated` when