fix(workouts): resolve duplicate clusters the same way on every surface

The cross-source picker anchors each duplicate cluster on its first start,
so which cluster a workout lands in depends on the workouts around it. The
detail route still read the old fixed five-minute clock slot, and every
other reader handed the picker only the rows of its own range. The detail
of a twin that straddled a slot boundary declared itself canonical while
the list hid it, and runs at 09:58, 10:02 and 10:06 were grouped one way by
a read starting at 09:55 and another by a read starting at 10:00.

Every reader that decides which workout is canonical (list, detail and its
previous session, activity calendar, day view, doctor report, Coach,
briefing and correlation blocks, strain, per-sport context, the activity
insight's history, the batch write) now reads through one windowed reader.
It reads a duplicate window past the end of the range and reaches back
until every sport in the range starts at a workout that is an anchor
whatever came before it, then picks over everything read and keeps the
rows that start inside the range. The per-sport baseline and the insight
history leave out the whole cluster of the current session instead of the
row alone, which left its twin behind as a session of its own. The batch
write clusters its rows together with the stored workouts around them and
drops only a row that loses to another row of the same batch.

A property test checks the windowed read against a whole-history read for
random dense histories and ranges; route tests pin list, detail, deep link,
previous session and the batch write against the database.
This commit is contained in:
Marc-André Bombeck committed 2026-10-22 18:21:02 +02:00
1 parent 3decab47a0
commit 466dfb3785
33 files changed
+1486 -472

No files matched your search

+5 -3
View File
@@ -4985,9 +4985,11 @@ paths:
- Measurements
summary: List workouts (v1.4.32)
description: "Paginated workout list with cross-source canonical-row dedup. The picker reads the per-user
source-priority ladder and buckets by `(startedAt 5 min slot, sportType)` — twin workouts (Apple Watch +
Withings ScanWatch) collapse to a single canonical row per cluster. Field names mirror the iOS handoff contract:
`distanceM`, `activeEnergyKcal`, `avgHr`, `maxHr`."
source-priority ladder and clusters workouts of one sport whose starts lie less than five minutes after the
cluster's first start — twin workouts (Apple Watch + Withings ScanWatch) collapse to a single canonical row per
cluster. A twin just outside `since` / `until` still decides which row inside the range is canonical, so the
list agrees with `canonicalId` on the detail. Field names mirror the iOS handoff contract: `distanceM`,
`activeEnergyKcal`, `avgHr`, `maxHr`."
parameters:
- in: query
name: limit
@@ -1,7 +1,7 @@
{
"$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
"count": 124,
"total": 146512,
"count": 123,
"total": 145651,
"entries": {
"app/api/dashboard/summary/route.ts": 1032,
"app/api/documents/inbound/route.ts": 811,
@@ -10,7 +10,7 @@
"app/api/measurements/route.ts": 938,
"app/api/medications/[id]/route.ts": 1281,
"app/api/medications/intake/bulk/route.ts": 1044,
"app/api/workouts/batch/route.ts": 971,
"app/api/workouts/batch/route.ts": 958,
"app/medications/page-client.tsx": 839,
"app/page-client.tsx": 2324,
"components/admin/backups-section.tsx": 1462,
@@ -44,7 +44,7 @@
"lib/ai/coach/coach-prose-grounding.ts": 1079,
"lib/ai/coach/persistence.ts": 1219,
"lib/ai/coach/results/metric-table-tool.ts": 1102,
"lib/ai/coach/snapshot.ts": 1362,
"lib/ai/coach/snapshot.ts": 1347,
"lib/ai/coach/stream-events.ts": 817,
"lib/ai/coach/tools/executor.ts": 1522,
"lib/ai/coach/turn/model.ts": 820,
@@ -65,7 +65,7 @@
"lib/daily/today-overview.ts": 838,
"lib/dashboard-layout.ts": 971,
"lib/dashboard/snapshot.ts": 1550,
"lib/day/records.ts": 1175,
"lib/day/records.ts": 1174,
"lib/doctor-report/collect.ts": 803,
"lib/export/backup-plan.ts": 853,
"lib/export/coach-backup.ts": 1070,
@@ -90,7 +90,6 @@
"lib/insights/mood-aggregates.ts": 1055,
"lib/insights/mood-status.ts": 814,
"lib/insights/status-cache.ts": 812,
"lib/insights/strain-score.ts": 832,
"lib/integrations/status.ts": 1445,
"lib/jobs/apple-health-import-worker.ts": 863,
"lib/jobs/backup-restore.ts": 888,
@@ -1,7 +1,7 @@
{
"$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
"count": 411,
"total": 148369,
"total": 148324,
"entries": {
"app/achievements/page.tsx :: AchievementsPage": 206,
"app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314,
@@ -64,8 +64,8 @@
"app/api/sync/changes/route.ts :: apiHandler(arg)": 427,
"app/api/user/ai-provider/route.ts :: apiHandler(arg) #2": 252,
"app/api/withings/callback/route.ts :: apiHandler(arg)": 247,
"app/api/workouts/[id]/route.ts :: apiHandler(arg)": 313,
"app/api/workouts/batch/route.ts :: postBatch": 722,
"app/api/workouts/[id]/route.ts :: apiHandler(arg)": 297,
"app/api/workouts/batch/route.ts :: postBatch": 709,
"app/auth/login/page.tsx :: LoginPage": 576,
"app/auth/register/page.tsx :: RegisterPage": 285,
"app/coach/conversations/page.tsx :: CoachConversationsBody": 249,
@@ -287,7 +287,7 @@
"lib/ai/anthropic-client.ts :: generateCompletion": 294,
"lib/ai/coach/cycle-snapshot.ts :: buildCycleSnapshotBlock": 218,
"lib/ai/coach/glp1-snapshot.ts :: buildGlp1SnapshotBlock": 218,
"lib/ai/coach/snapshot.ts :: buildCoachSnapshotImpl": 898,
"lib/ai/coach/snapshot.ts :: buildCoachSnapshotImpl": 882,
"lib/ai/coach/tools/loop.ts :: runCoachToolLoop": 438,
"lib/ai/coach/turn/model.ts :: runTurnModel": 560,
"lib/ai/coach/turn/pipeline.ts :: runTurn": 274,
+2 -2
View File
@@ -1,7 +1,7 @@
{
"$comment": "as never casts in tests, per test file under src/. Maintained by src/__tests__/test-realism-ratchet.test.ts.",
"count": 524,
"total": 5496,
"total": 5495,
"entries": {
"__tests__/api/auth/login/mfa-branch.test.ts": 10,
"__tests__/api/auth/mfa/verify.test.ts": 9,
@@ -253,7 +253,7 @@
"app/api/withings/credentials/__tests__/route.test.ts": 4,
"app/api/withings/webhook/__tests__/route.test.ts": 26,
"app/api/withings/webhook/[token]/__tests__/route.test.ts": 8,
"app/api/workouts/__tests__/batch-create.test.ts": 14,
"app/api/workouts/__tests__/batch-create.test.ts": 13,
"app/api/workouts/__tests__/canonical-dedup.test.ts": 20,
"app/api/workouts/batch/__tests__/unstable-external-id.test.ts": 4,
"app/c/[token]/__tests__/report-download.test.ts": 11,
@@ -1,7 +1,7 @@
{
"$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.",
"count": 2959,
"total": 10521,
"total": 10520,
"entries": {
"__tests__/account-payload-consumer-guard.test.ts": 2,
"__tests__/acting-account-boundary-guard.test.ts": 7,
@@ -1710,7 +1710,7 @@
"lib/ai/coach/snapshot-cache.ts": 2,
"lib/ai/coach/snapshot-prelude.ts": 17,
"lib/ai/coach/snapshot-series.ts": 5,
"lib/ai/coach/snapshot.ts": 42,
"lib/ai/coach/snapshot.ts": 41,
"lib/ai/coach/source-measurement-types.ts": 1,
"lib/ai/coach/suggest-action.ts": 1,
"lib/ai/coach/suggest-gate.ts": 1,
@@ -31,11 +31,21 @@ const SRC = join(process.cwd(), "src");
const PICKER = "lib/measurements/pick-canonical-workout-rows.ts";
/**
* Every production call site of the picker, frozen.
* The only module that hands rows to the picker: the windowed reader, which
* picks over rows it read with the duplicate-window margin. Every surface goes
* through it, and its output types make a read without the margin a compile
* error. The strain engine's per-cluster impulse sum lives there too and is
* handed the rows of the engine's own windowed read.
*/
const PICKER_CALLERS = ["lib/workouts/canonical-read.ts"];
/**
* Every production read that decides which workout is canonical, frozen.
*
* Adding a surface that collapses workouts is a deliberate act: it must fetch
* the identity fields or it silently disagrees with every other surface. Adding
* the file here is the acknowledgement.
* Adding a surface that collapses workouts is a deliberate act: it must read
* through `readWorkoutsAroundRange` and fetch the identity fields, or it
* silently disagrees with every other surface. Adding the file here is the
* acknowledgement. The value is the file that owns the read's `select`.
*/
const IDENTITY_OWNER: Record<string, string> = {
"app/api/workouts/[id]/route.ts": "app/api/workouts/[id]/route.ts",
@@ -49,15 +59,15 @@ const IDENTITY_OWNER: Record<string, string> = {
// The activity calendar counts workouts per day; a session two devices
// recorded must count once there, as it does in the list.
"lib/activity/calendar-read.ts": "lib/activity/calendar-read.ts",
// The only indirection in the set. The Coach block is handed rows it does not
// read itself; the snapshot builder fetches them in parallel with every other
// block's data. Recording the owner here rather than assuming the call site
// owns its own select is the whole point of this map.
"lib/ai/coach/snapshot-blocks/workouts-block.ts": "lib/ai/coach/snapshot.ts",
// The write path: the batch route runs the same picker over the rows it is
// about to insert, so a duplicate inside one batch is dropped by the rule
// the list later applies.
"app/api/workouts/batch/route.ts": "app/api/workouts/batch/route.ts",
// The Coach block is handed rows it does not read itself; the snapshot
// builder fetches them through this read in parallel with every other
// block's data.
"lib/ai/coach/snapshot-blocks/workouts-read.ts":
"lib/ai/coach/snapshot-blocks/workouts-read.ts",
// The write path: the batch route clusters the rows it is about to insert
// together with the stored rows around them, so a duplicate is dropped by
// the rule the list later applies.
"lib/workouts/batch-dedup.ts": "lib/workouts/batch-dedup.ts",
// The day-total training impulse, the briefing's workout tally and the day
// view's workout section: each counted a session synced by two services
// twice while the list showed it once.
@@ -72,10 +82,16 @@ const IDENTITY_OWNER: Record<string, string> = {
/**
* How a call site proves its rows carry the duration. A read selects the
* column; the batch route maps it off the row it computed for the insert.
* column; the batch route also maps it off the row it computed for the insert.
*/
const IDENTITY_PROOF: Record<string, string> = {
"app/api/workouts/batch/route.ts": "durationSec: p.row.durationSec",
const IDENTITY_PROOF: Record<string, string[]> = {
// These two select `WORKOUT_PICK_SELECT`, whose duration is checked below.
"lib/day/records.ts": ["select: WORKOUT_PICK_SELECT"],
"lib/insights/correlation-channel-series.ts": ["select: WORKOUT_PICK_SELECT"],
"lib/workouts/batch-dedup.ts": [
"durationSec: true",
"durationSec: p.row.durationSec",
],
};
const EXPECTED_CALL_SITES = Object.keys(IDENTITY_OWNER).sort();
@@ -93,17 +109,13 @@ function read(rel: string): string {
}
/**
* Lines that actually invoke the picker. Comment lines are excluded: a
* docblock that names the picker is documentation rather than a call.
* Lines that actually invoke `call`. Comment lines are excluded: a docblock
* that names the function is documentation rather than a call.
*/
function invokesPicker(source: string): boolean {
function invokes(source: string, call: string): boolean {
return source
.split("\n")
.some(
(line) =>
line.includes("pickCanonicalWorkoutRows(") &&
!/^\s*(\*|\/\/|\/\*)/.test(line),
);
.some((line) => line.includes(call) && !/^\s*(\*|\/\/|\/\*)/.test(line));
}
describe("the picker's call-site set is frozen", () => {
@@ -114,13 +126,29 @@ describe("the picker's call-site set is frozen", () => {
expect(sourceFiles().length).toBeGreaterThan(1500);
});
it("has exactly the call sites this file acknowledges", () => {
it("hands rows to the picker only from the windowed reader", () => {
const found = sourceFiles()
.filter((rel) => rel !== PICKER)
.filter((rel) => invokesPicker(read(rel)));
.filter((rel) => invokes(read(rel), "pickCanonicalWorkoutRows("));
expect(found).toEqual(PICKER_CALLERS);
});
it("has exactly the windowed reads this file acknowledges", () => {
const found = sourceFiles()
.filter((rel) => rel !== "lib/workouts/canonical-read.ts")
.filter((rel) => invokes(read(rel), "readWorkoutsAroundRange("));
expect(found).toEqual(EXPECTED_CALL_SITES);
});
it("scores strain only on the sessions of the windowed range", () => {
// The strain engine sums impulses over the rows of its windowed read; it
// must keep the sessions its range owns, not every cluster the margin read.
expect(read("lib/insights/strain-score.ts")).toContain(
"(at) => startsInRange(workouts.range, at)",
);
});
});
describe("every call site fetches the session identity", () => {
@@ -130,9 +158,20 @@ describe("every call site fetches the session identity", () => {
// rows came from. Coarse in the safe direction — it can only fail to catch
// a second unrelated select in the same file, never pass a caller that
// fetches no duration at all.
expect(read(IDENTITY_OWNER[rel])).toContain(
IDENTITY_PROOF[rel] ?? "durationSec: true",
for (const proof of IDENTITY_PROOF[rel] ?? ["durationSec: true"]) {
expect(read(IDENTITY_OWNER[rel])).toContain(proof);
}
});
});
describe("the shared pick select carries the identity", () => {
it("selects durationSec", () => {
const reader = read("lib/workouts/canonical-read.ts");
const select = reader.slice(
reader.indexOf("export const WORKOUT_PICK_SELECT"),
reader.indexOf("} as const;"),
);
expect(select).toContain("durationSec: true");
});
});
+104 -81
View File
@@ -6,9 +6,9 @@
* canonical envelope per workout. The endpoint:
*
* - Returns the canonical row when the requested id wins its
* `(startedAt slot, sportType)` cluster against the user's
* source-priority ladder. The picker reuses
* `pickCanonicalWorkoutRows()` from v1.4.30 so the dedup contract
* duplicate cluster against the user's source-priority ladder. The
* cluster is resolved through the same windowed read the list uses
* (`src/lib/workouts/canonical-read.ts`), so the dedup contract
* stays in sync with the list endpoint.
* - When the requested id is a non-canonical twin (e.g. the user
* opened a deep-link to a Withings row but the cluster's winner is
@@ -40,7 +40,14 @@ import {
lockPersonalRecordsForUser,
workoutPrSlotsForSport,
} from "@/lib/personal-records/pr-detection-worker";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalClusterOf,
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
type CanonicalReadRow,
type WorkoutBoundedRead,
type WorkoutStartBounds,
} from "@/lib/workouts/canonical-read";
import { requireModuleEnabled } from "@/lib/modules/gate";
import { aiCapabilityToServe } from "@/lib/ai/capabilities/gate";
import { getAgeFromDateOfBirth } from "@/lib/analytics/pulse-targets";
@@ -60,6 +67,38 @@ import { userDayKey } from "@/lib/tz/format";
type RouteParams = { params: Promise<{ id: string }> };
/** How many earlier rows one step of the previous-session search reads. */
const PREVIOUS_SESSION_BATCH = 20;
/** Steps before the search gives up; each reaches past the last one. */
const PREVIOUS_SESSION_MAX_STEPS = 5;
/**
* The newest canonical workout of the sport that starts before `before` (the
* first start of the current session's cluster). A batch of earlier rows names
* the span to resolve; the span is read through the windowed reader so its
* twins are resolved the way the list resolves them. A span whose rows are all
* hidden by a twin further back moves the search back one batch.
*/
async function previousCanonicalWorkoutId<T extends CanonicalReadRow>(
read: WorkoutBoundedRead<T>,
latestStartsBefore: (to: Date, take: number) => Promise<Date[]>,
before: Date,
priorityJson: unknown,
): Promise<string | null> {
let to = before;
for (let step = 0; step < PREVIOUS_SESSION_MAX_STEPS; step += 1) {
const earlier = await latestStartsBefore(to, PREVIOUS_SESSION_BATCH);
if (earlier.length === 0) return null;
const from = earlier[earlier.length - 1]!;
const windowed = await readWorkoutsAroundRange({ from, to: before }, read);
const canonical = canonicalWorkoutsInRange(windowed, priorityJson);
const newest = canonical[canonical.length - 1];
if (newest) return newest.id;
to = from;
}
return null;
}
export const GET = apiHandler(
async (request: NextRequest, { params }: RouteParams) => {
const { user } = await requireAuth();
@@ -113,38 +152,38 @@ export const GET = apiHandler(
return apiError("Workout not found", 404);
}
// Resolve the canonical winner for the row's cluster. The picker
// reads every workout in the same 5-minute slot and sport-type
// bucket; in practice that's ≤ 4 rows (Apple + Withings + the two
// legacy sources). The query is bounded so the round-trip is cheap.
const slotMs = 5 * 60 * 1000;
const slotStart = new Date(
Math.floor(row.startedAt.getTime() / slotMs) * slotMs,
// Resolve the canonical winner for the row's cluster. The read goes
// through the same windowed reader as the list, so the cluster drawn here
// is the one the list draws: a twin that straddles a clock slot boundary,
// or one whose cluster begins with an earlier workout, resolves to the row
// the list shows.
const readSameSport = (bounds: WorkoutStartBounds) =>
prisma.workout.findMany({
where: {
userId: user.id,
sportType: row.sportType,
startedAt: bounds,
},
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
source: true,
startedAt: true,
sportType: true,
// Part of the canonical session identity consulted by the picker —
// without it a HealthKit re-send of this very workout stays in the
// cluster and the deep link can resolve to the duplicate instead of
// the row the list shows.
durationSec: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
},
});
const around = await readWorkoutsAroundRange(
{ from: row.startedAt, to: row.startedAt, inclusiveEnd: true },
readSameSport,
);
const slotEnd = new Date(slotStart.getTime() + slotMs);
const clusterRows = await prisma.workout.findMany({
where: {
userId: user.id,
sportType: row.sportType,
startedAt: { gte: slotStart, lt: slotEnd },
},
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
source: true,
startedAt: true,
sportType: true,
// Part of the canonical session identity consulted by
// `pickCanonicalWorkoutRows` below — without it a HealthKit re-send of
// this very workout stays in the cluster and the deep link can resolve
// to the duplicate instead of the row the list shows.
durationSec: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
},
});
const userRow = await prisma.user.findUnique({
where: { id: user.id },
@@ -158,17 +197,15 @@ export const GET = apiHandler(
const timezone = await resolveUserTimezone(user.id);
const dayKey = userDayKey(row.startedAt, timezone);
const canonicalCluster = pickCanonicalWorkoutRows(
clusterRows,
const cluster = canonicalClusterOf(
around,
row.id,
userRow?.sourcePriorityJson ?? null,
);
// The canonical pick may carry multiple rows (same winning source);
// pick the closest in time to the requested row so deep-link
// redirects land on the most similar twin.
const canonical =
canonicalCluster.find((c) => c.id === row.id) ??
canonicalCluster[0] ??
null;
// The cluster's canonical rows, the requested row first when it is one of
// them and otherwise the nearest in time, so a deep link to a hidden twin
// lands on the most similar row.
const canonical = cluster?.canonical[0] ?? null;
const effectiveAvgHeartRate = canonical?.avgHeartRate ?? row.avgHeartRate;
const effectiveMaxHeartRate = canonical?.maxHeartRate ?? row.maxHeartRate;
@@ -233,43 +270,29 @@ export const GET = apiHandler(
// The session before this one in the same sport. "How does this compare
// to last time?" is the most likely next question after reading a
// workout, and today it costs a trip back to the list plus a scan.
// Scoped to rows that start before this cluster's slot, so a twin of THIS
// session can never be offered as the previous one, and run through the
// same canonical picker so the link lands on the row the list shows.
const previousRows = await prisma.workout.findMany({
where: {
userId: user.id,
sportType: row.sportType,
startedAt: { lt: slotStart },
},
orderBy: [{ startedAt: "desc" }, { id: "desc" }],
// Enough to survive a run of same-session twins and re-sends without
// reading the whole history; the picker collapses them to a handful.
take: 20,
select: {
id: true,
source: true,
startedAt: true,
sportType: true,
durationSec: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
},
});
// The picker takes rows in ascending order and does not promise to hand
// them back sorted, so pick the newest survivor explicitly.
const previousWorkoutId =
pickCanonicalWorkoutRows(
[...previousRows].reverse(),
userRow?.sourcePriorityJson ?? null,
).reduce<{ id: string; startedAt: Date } | null>(
(newest, candidate) =>
newest === null || candidate.startedAt > newest.startedAt
? candidate
: newest,
null,
)?.id ?? null;
// Scoped to rows that start before this session's cluster, so a twin of
// THIS session can never be offered as the previous one, and resolved
// through the same windowed read so the link lands on the row the list
// shows.
const clusterStart = cluster?.members[0]?.startedAt ?? row.startedAt;
const previousWorkoutId = await previousCanonicalWorkoutId(
readSameSport,
async (to, take) =>
(
await prisma.workout.findMany({
where: {
userId: user.id,
sportType: row.sportType,
startedAt: { lt: to },
},
orderBy: [{ startedAt: "desc" }, { id: "desc" }],
take,
select: { startedAt: true },
})
).map((earlier) => earlier.startedAt),
clusterStart,
userRow?.sourcePriorityJson ?? null,
);
// A route whose track cannot be read (a sealed value that no longer
// opens, or an old row stored as a JSON null) is reported as no route at
@@ -128,6 +128,19 @@ function mockInsertReturning(
}) => (args.where === undefined && args.select?.id ? rows : [])) as never);
}
/**
* Rows the pre-flight `(source, externalId)` probe finds. The probe is the
* read with an `OR` of keys; the windowed read of the stored workouts around
* the batch (a `startedAt` range) finds nothing here.
*/
function mockExistingByKey(
rows: Array<{ source: string; externalId: string }>,
) {
vi.mocked(prisma.workout.findMany).mockImplementation((async (args: {
where?: { OR?: unknown };
}) => (args.where?.OR ? rows : [])) as never);
}
function validWorkout(
externalId: string,
overrides: Record<string, unknown> = {},
@@ -345,9 +358,9 @@ describe("POST /api/workouts/batch — per-entry status envelope", () => {
});
it("flags entries already in the DB as duplicate without re-inserting", async () => {
vi.mocked(prisma.workout.findMany).mockResolvedValue([
mockExistingByKey([
{ source: "APPLE_HEALTH", externalId: "uuid-existing" },
] as never);
]);
vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 1 });
const res = await POST(
makeRequest({
@@ -825,9 +838,7 @@ describe("POST /api/workouts/batch — PR detection enqueue (v1.4.25 W16c)", ()
// stays 0 but duplicateCount becomes 1. The hook fires anyway —
// the detector is cheap and a duplicate batch can still reveal a
// PR if the row was written by another path between dispatches.
vi.mocked(prisma.workout.findMany).mockResolvedValue([
{ source: "APPLE_HEALTH", externalId: "uuid-dup" } as never,
]);
mockExistingByKey([{ source: "APPLE_HEALTH", externalId: "uuid-dup" }]);
vi.mocked(prisma.workout.createMany).mockResolvedValue({ count: 0 });
const res = await POST(
makeRequest({ workouts: [validWorkout("uuid-dup")] }),
@@ -358,9 +358,11 @@ describe("GET /api/workouts — canonical dedup", () => {
where: {
userId: "user-1",
sportType: "running",
// The range plus one duplicate window on either side, so a twin
// just outside the filter still decides which row is canonical.
startedAt: {
gte: new Date("2026-01-01T00:00:00.000Z"),
lte: new Date("2026-01-02T00:00:00.000Z"),
gte: new Date("2025-12-31T23:55:00.000Z"),
lt: new Date("2026-01-02T00:05:00.000Z"),
},
},
}),
+10 -23
View File
@@ -92,11 +92,11 @@ import {
UNSTABLE_EXTERNAL_ID_REASON,
type UnstableExternalIdShape,
} from "@/lib/validations/external-id";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import { batchWorkoutIndicesToKeep } from "@/lib/workouts/batch-dedup";
import { WORKOUTS_WRITE_SCOPE } from "@/lib/workouts/scopes";
import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source";
import { encryptRouteGeometry } from "@/lib/workouts/route-geometry-cipher";
import { Prisma, type MeasurementSource } from "@/generated/prisma/client";
import { Prisma } from "@/generated/prisma/client";
// v1.4.25 — push-suppression threshold for workout PRs. A batch
// larger than this fires the detection job with `silent: true` so a
@@ -508,29 +508,16 @@ async function postBatch(request: NextRequest): Promise<Response> {
});
}
// Write-time cross-source dedup through the one workout picker the reads
// use, with the same source ladder: one logical workout from two sources
// in one batch (or one HealthKit session re-sent under a new id) lands
// once. Rows the picker hides surface to the client as `duplicate`, so
// the sync cursor advances past them like the externalId case below.
const canonicalPrepared = pickCanonicalWorkoutRows(
prepared.map((p) => ({
index: p.index,
userId: p.row.userId,
sportType: p.row.sportType,
startedAt: p.row.startedAt as Date,
durationSec: p.row.durationSec,
// Zod's `measurementSourceEnum.optional().default("MANUAL")`
// guarantees `source` is set post-parse, but Prisma's
// `WorkoutCreateManyInput` types it optional.
source: (p.row.source ?? "MANUAL") as MeasurementSource,
avgHeartRate: p.row.avgHeartRate ?? null,
maxHeartRate: p.row.maxHeartRate ?? null,
metadata: p.row.metadata ?? null,
})),
// Write-time cross-source dedup through the one workout clustering the
// reads use (see `batchWorkoutIndicesToKeep`): one logical workout from two
// sources in one batch, or one HealthKit session re-sent under a new id,
// lands once. Rows hidden here surface to the client as `duplicate`, so the
// sync cursor advances past them like the externalId case below.
const survivingIndices = await batchWorkoutIndicesToKeep(
user.id,
prepared,
userPriority?.sourcePriorityJson ?? null,
);
const survivingIndices = new Set(canonicalPrepared.map((r) => r.index));
const droppedByWriteDedup: number[] = [];
for (const p of prepared) {
if (!survivingIndices.has(p.index)) {
+24 -16
View File
@@ -15,7 +15,10 @@
*/
import type { MeasurementType } from "@/generated/prisma/client";
import { prisma } from "@/lib/db";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
} from "@/lib/workouts/canonical-read";
import { seriesRowsFrom } from "@/lib/measurements/series-canonical";
import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
import { startOfLocalDayKey } from "@/lib/tz/local-day";
@@ -118,26 +121,31 @@ export async function readActivityDays(args: {
),
),
includeWorkouts
? prisma.workout.findMany({
where: { userId, startedAt: { gte: from, lt: to } },
select: {
startedAt: true,
durationSec: true,
sportType: true,
source: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
},
orderBy: { startedAt: "asc" },
})
: Promise.resolve([]),
? readWorkoutsAroundRange({ from, to }, (bounds) =>
prisma.workout.findMany({
where: { userId, startedAt: bounds },
select: {
id: true,
startedAt: true,
durationSec: true,
sportType: true,
source: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
},
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
}),
)
: Promise.resolve(null),
]);
const loadIndex = loads.findIndex((m) => m.size > 0);
const load = loadIndex >= 0 ? loads[loadIndex] : null;
const workouts = pickCanonicalWorkoutRows(workoutRows, priorityJson);
const workouts = workoutRows
? canonicalWorkoutsInRange(workoutRows, priorityJson)
: [];
const workoutsByDay = new Map<
string,
{ minutes: number; count: number; bySport: Record<string, number> }
@@ -372,9 +372,12 @@ describe("buildCoachSnapshot — Apple Health additive metrics (v1.4.23)", () =>
// collapse, so a dual-device user's Coach prompt double-counted the
// session and inflated `perSport`/`totalInWindow` vs. the app. Pin the fix.
it("collapses a same-session cross-source workout pair to one, matching the UI", async () => {
const sharedStart = new Date("2026-06-03T06:30:00Z");
// Inside the 30-day window: the block keeps only the sessions that
// start in it.
const sharedStart = new Date(Date.now() - 3 * 24 * 60 * 60 * 1000);
const workouts = [
{
id: "w-apple",
sportType: "RUNNING",
startedAt: sharedStart,
durationSec: 1800,
@@ -385,6 +388,7 @@ describe("buildCoachSnapshot — Apple Health additive metrics (v1.4.23)", () =>
source: "APPLE_HEALTH",
},
{
id: "w-withings",
sportType: "RUNNING",
// 90 s apart — same 5-minute bucket the picker uses.
startedAt: new Date(sharedStart.getTime() + 90_000),
@@ -12,18 +12,23 @@ import type { UnitPreference } from "@/lib/measurements/display-transform";
function build(unitPreference: UnitPreference) {
const snapshot: Record<string, unknown> = {};
buildWorkoutsBlock({
workoutRows: [
{
sportType: "running",
startedAt: new Date("2026-09-20T07:00:00Z"),
durationSec: 1800,
totalEnergyKcal: 300,
totalDistanceM: 5000,
avgHeartRate: 150,
maxHeartRate: 170,
source: "APPLE_HEALTH",
},
],
workoutRows: {
range: { from: null, to: null },
__windowed: true,
rows: [
{
id: "w1",
sportType: "running",
startedAt: new Date("2026-09-20T07:00:00Z"),
durationSec: 1800,
totalEnergyKcal: 300,
totalDistanceM: 5000,
avgHeartRate: 150,
maxHeartRate: 170,
source: "APPLE_HEALTH",
},
],
},
sourcePriorityJson: null,
userTz: "UTC",
snapshot,
@@ -25,7 +25,10 @@ import {
getQuantityTransform,
type UnitPreference,
} from "@/lib/measurements/display-transform";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
type WindowedWorkouts,
} from "@/lib/workouts/canonical-read";
import type { MeasurementSource } from "@/generated/prisma/client";
import { annotate } from "@/lib/logging/context";
import { tzDayKey, tzWeekday } from "../snapshot-series";
@@ -43,7 +46,8 @@ import type {
const WORKOUT_RECENT_CAP = 15;
interface WorkoutsBlockContext {
workoutRows: Array<{
workoutRows: WindowedWorkouts<{
id: string;
sportType: string;
startedAt: Date;
durationSec: number;
@@ -85,7 +89,7 @@ export function buildWorkoutsBlock(ctx: Readonly<WorkoutsBlockContext>): void {
// `startedAt` ASC (its documented contract) — the builder wants
// most-recent-first for the capped `recent` list, so re-establish
// DESC order after the pick, same as `GET /api/workouts` does.
const workoutRows = pickCanonicalWorkoutRows(
const workoutRows = canonicalWorkoutsInRange(
rawWorkoutRows,
sourcePriorityJson ?? null,
).sort((a, b) => b.startedAt.getTime() - a.startedAt.getTime());
@@ -0,0 +1,30 @@
/**
* The workout rows the Coach's workouts block narrates, read through the
* windowed reader so a session two sources recorded collapses to the row the
* workout list shows, also at the edge of the Coach's window.
*/
import { prisma } from "@/lib/db";
import { readWorkoutsAroundRange } from "@/lib/workouts/canonical-read";
export function readCoachWorkoutRows(userId: string, since: Date) {
return readWorkoutsAroundRange({ from: since, to: null }, (bounds) =>
prisma.workout.findMany({
where: { userId, startedAt: bounds },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
sportType: true,
startedAt: true,
durationSec: true,
totalEnergyKcal: true,
totalDistanceM: true,
avgHeartRate: true,
maxHeartRate: true,
// Needed by the canonical pick in the block builder to collapse a
// dual-source session (e.g. Apple Watch + WHOOP recording the same
// run) to one, matching `GET /api/workouts`.
source: true,
},
}),
);
}
+2 -17
View File
@@ -79,6 +79,7 @@ import {
buildCoarseTimelineTail,
type CoarseTimelineTail,
} from "./snapshot-series";
import { readCoachWorkoutRows } from "./snapshot-blocks/workouts-read";
import { buildWorkoutsBlock } from "./snapshot-blocks/workouts-block";
import {
buildGlucoseBlock,
@@ -567,23 +568,7 @@ async function buildCoachSnapshotImpl(
? readSleepStageRows(userId, sleepRhythmCutoff)
: null;
const workoutRowsPromise = sources.has("workouts")
? prisma.workout.findMany({
where: { userId, startedAt: { gte: additiveCutoff("workouts") } },
orderBy: { startedAt: "desc" },
select: {
sportType: true,
startedAt: true,
durationSec: true,
totalEnergyKcal: true,
totalDistanceM: true,
avgHeartRate: true,
maxHeartRate: true,
// v1.30.4 — needed by `pickCanonicalWorkoutRows` in the block
// builder to collapse a dual-source session (e.g. Apple Watch +
// WHOOP recording the same run) to one, matching `GET /api/workouts`.
source: true,
},
})
? readCoachWorkoutRows(userId, additiveCutoff("workouts"))
: null;
// v1.17.0 — the glucose CLINICAL panel is a fixed 30-day clinical artifact,
// identical to the one the insights panel + doctor report render. It must NOT
+16 -17
View File
@@ -42,7 +42,11 @@ import { effectiveMoodTz } from "@/lib/mood/date-key";
import { dateOnlyKey, dayKeyAsUtcMidnight } from "@/lib/tz/date-only";
import { daysBetweenDateKeys, shiftDateKey, userDayKey } from "@/lib/tz/format";
import { visitPath } from "@/lib/encounters/visit-links";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
WORKOUT_PICK_SELECT,
} from "@/lib/workouts/canonical-read";
/** Everything a section reader needs about the day. */
export interface DayFrame {
@@ -906,28 +910,23 @@ async function assessmentsPart(frame: DayFrame): Promise<SectionPart> {
async function workoutsPart(frame: DayFrame): Promise<SectionPart> {
const [stored, account] = await Promise.all([
prisma.workout.findMany({
where: {
userId: frame.userId,
startedAt: { gte: frame.dayStart, lt: frame.dayEnd },
},
select: {
id: true,
source: true,
sportType: true,
startedAt: true,
durationSec: true,
},
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
}),
// The windowed reader orders the rows itself.
readWorkoutsAroundRange(
{ from: frame.dayStart, to: frame.dayEnd },
(bounds) =>
prisma.workout.findMany({
where: { userId: frame.userId, startedAt: bounds },
select: WORKOUT_PICK_SELECT,
}),
),
prisma.user.findUnique({
where: { id: frame.userId },
select: { sourcePriorityJson: true },
}),
]);
// One session synced by two services is one workout on the day, as in the
// workout list it links to.
const rows = pickCanonicalWorkoutRows(
// workout list it links to, a twin across midnight included.
const rows = canonicalWorkoutsInRange(
stored,
account?.sourcePriorityJson ?? null,
);
+23 -15
View File
@@ -11,7 +11,10 @@
* session two sources reported counts once.
*/
import { prisma } from "@/lib/db";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
} from "@/lib/workouts/canonical-read";
import type { DoctorReportWorkouts } from "@/lib/doctor-report-types";
/**
@@ -29,20 +32,25 @@ export async function loadReportWorkouts(params: {
sourcePriorityJson: unknown;
}): Promise<DoctorReportWorkouts | null> {
const { userId, start, end, days, sourcePriorityJson } = params;
const rows = await prisma.workout.findMany({
where: { userId, startedAt: { gte: start, lte: end } },
orderBy: { startedAt: "asc" },
select: {
sportType: true,
startedAt: true,
durationSec: true,
source: true,
totalDistanceM: true,
totalEnergyKcal: true,
avgHeartRate: true,
},
});
const canonical = pickCanonicalWorkoutRows(rows, sourcePriorityJson);
const rows = await readWorkoutsAroundRange(
{ from: start, to: end, inclusiveEnd: true },
(bounds) =>
prisma.workout.findMany({
where: { userId, startedAt: bounds },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
sportType: true,
startedAt: true,
durationSec: true,
source: true,
totalDistanceM: true,
totalEnergyKcal: true,
avgHeartRate: true,
},
}),
);
const canonical = canonicalWorkoutsInRange(rows, sourcePriorityJson);
return summariseReportWorkouts(canonical, days);
}
@@ -9,6 +9,7 @@
* - the per-day idempotency key + insufficient-data gate.
*/
import { describe, it, expect, vi, beforeEach } from "vitest";
import { canonicalWorkoutTrimps } from "@/lib/workouts/canonical-read";
/**
* The post-mutation tail reads the real `@/lib/db` client, not the fake this
@@ -32,7 +33,6 @@ import {
percentile,
resolvePersonalReference,
persistStrainScore,
canonicalWorkoutTrimps,
canonicalDayActiveEnergy,
STRAIN_ACTIVE_ENERGY_REFERENCE,
STRAIN_SCORE_EXTERNAL_ID_PREFIX,
+14 -14
View File
@@ -76,7 +76,11 @@ import {
userDayKey,
} from "@/lib/tz/format";
import { MIN_PAIRED_N } from "@/lib/insights/correlations";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
WORKOUT_PICK_SELECT,
} from "@/lib/workouts/canonical-read";
import { probeRollupCoverage } from "@/lib/rollups/measurement-coverage";
import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
import type {
@@ -1257,18 +1261,14 @@ export async function fetchTrainingSeries(
since: Date,
): Promise<NamedSeries[]> {
const [workouts, account, trimp] = await Promise.all([
prisma.workout.findMany({
where: { userId, startedAt: { gte: since } },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
take: WORKOUT_READ_CAP,
select: {
id: true,
source: true,
sportType: true,
startedAt: true,
durationSec: true,
},
}),
readWorkoutsAroundRange({ from: since, to: null }, (bounds) =>
prisma.workout.findMany({
where: { userId, startedAt: bounds },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
take: WORKOUT_READ_CAP,
select: WORKOUT_PICK_SELECT,
}),
),
prisma.user.findUnique({
where: { id: userId },
select: { sourcePriorityJson: true },
@@ -1281,7 +1281,7 @@ export async function fetchTrainingSeries(
select: { day: true, dayTrimp: true },
}),
]);
const sessions = pickCanonicalWorkoutRows(
const sessions = canonicalWorkoutsInRange(
workouts,
account?.sourcePriorityJson ?? null,
);
+20 -14
View File
@@ -28,7 +28,10 @@ import {
readRollupBuckets,
ROLLUP_FOLD_WINDOW_MS,
} from "@/lib/rollups/measurement-rollups";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
} from "@/lib/workouts/canonical-read";
import { deriveBucketedTypes } from "@/lib/signals/adapters/correlation";
import type {
AggregatedFeatures,
@@ -537,18 +540,21 @@ export async function readWorkoutsBlock(
): Promise<AggregatedFeatures["workouts"] | undefined> {
const since = new Date(now - WORKOUT_WINDOW_DAYS * MS_PER_DAY);
const [stored, account] = await Promise.all([
prisma.workout.findMany({
where: { userId, startedAt: { gte: since } },
orderBy: [{ startedAt: "desc" }, { id: "asc" }],
take: 2000,
select: {
source: true,
sportType: true,
startedAt: true,
durationSec: true,
totalDistanceM: true,
},
}),
readWorkoutsAroundRange({ from: since, to: null }, (bounds) =>
prisma.workout.findMany({
where: { userId, startedAt: bounds },
orderBy: [{ startedAt: "desc" }, { id: "asc" }],
take: 2000,
select: {
id: true,
source: true,
sportType: true,
startedAt: true,
durationSec: true,
totalDistanceM: true,
},
}),
),
prisma.user.findUnique({
where: { id: userId },
select: { sourcePriorityJson: true },
@@ -558,7 +564,7 @@ export async function readWorkoutsBlock(
// workout: the same collapse the workout list applies, so the count and the
// minutes here match what the person sees there. The picker returns its rows
// oldest first; the block reads the newest at index 0.
const rows = pickCanonicalWorkoutRows(
const rows = canonicalWorkoutsInRange(
stored,
account?.sourcePriorityJson ?? null,
).reverse();
+26 -96
View File
@@ -73,11 +73,11 @@ import {
} from "@/lib/insights/score-row";
import { shiftDateKey } from "@/lib/tz/format";
import {
pickCanonicalWorkoutRows,
workoutDuplicateClusterKeys,
workoutSourceLadder,
type WorkoutPickerRow,
} from "@/lib/measurements/pick-canonical-workout-rows";
canonicalWorkoutTrimps,
readWorkoutsAroundRange,
rowsInRange,
startsInRange,
} from "@/lib/workouts/canonical-read";
import {
pickCanonicalSourceRows,
type SourcePickerRow,
@@ -433,19 +433,23 @@ export async function computeStrainScore(
// cross-source picker needs: one run recorded by the watch and synced again
// through a second service is one training impulse, not two.
const [workouts, account] = await Promise.all([
prisma.workout.findMany({
where: { userId, startedAt: { gte: dayStart, lt: dayEnd } },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
source: true,
sportType: true,
startedAt: true,
durationSec: true,
samples: { select: { samples: true } },
},
take: 100,
}),
// Read through the windowed reader: a session whose twin lies across
// midnight is counted once, on the day the list shows it.
readWorkoutsAroundRange({ from: dayStart, to: dayEnd }, (bounds) =>
prisma.workout.findMany({
where: { userId, startedAt: bounds },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
source: true,
sportType: true,
startedAt: true,
durationSec: true,
samples: { select: { samples: true } },
},
take: 100,
}),
),
prisma.user.findUnique({
where: { id: userId },
select: { sourcePriorityJson: true },
@@ -465,7 +469,7 @@ export async function computeStrainScore(
let dayTrimp = 0;
let workoutsWithSeries = 0;
if (canTrimp) {
const trimpOf = (w: (typeof workouts)[number]): number => {
const trimpOf = (w: (typeof workouts.rows)[number]): number => {
const raw = w.samples?.samples;
if (!Array.isArray(raw)) return 0;
return banisterTrimp(
@@ -476,9 +480,10 @@ export async function computeStrainScore(
);
};
const sessions = canonicalWorkoutTrimps(
workouts,
workouts.rows,
account?.sourcePriorityJson ?? null,
trimpOf,
(at) => startsInRange(workouts.range, at),
);
for (const trimp of sessions) {
dayTrimp += trimp;
@@ -535,7 +540,7 @@ export async function computeStrainScore(
return {
score: null,
reason:
workouts.length > 0 && !canTrimp
rowsInRange(workouts).length > 0 && !canTrimp
? "insufficient_profile"
: "insufficient_inputs",
dayTrimp: 0,
@@ -572,81 +577,6 @@ export function canonicalDayActiveEnergy<
return best;
}
/**
* One training impulse per workout SESSION, not per stored row.
*
* The same session can be stored more than once: the watch writes it to Apple
* Health and a second service (Strava, WHOOP, a ring) syncs it again, or
* HealthKit re-sends it under a new identifier. The workout list collapses
* those through `pickCanonicalWorkoutRows`; summing every row here counted the
* session twice in the day's TRIMP while the list showed it once.
*
* The picker chooses the row by source order, which says nothing about heart
* rate: the winning row can be one without a series while its dropped twin
* carries one. So the impulse of a slot is the sum over its picked rows, and
* only when that is zero the largest impulse among the slot's dropped rows.
*
* The picker keeps every row of a slot none of whose sources is on the
* ladder, so the picked rows of one slot can come from several sources. Only
* one source's rows count: the best-ranked on the ladder, a source off the
* ladder ranking below every listed one, and among sources off the ladder
* the one with the largest impulse. Rows of that one source still add up
* (two distinct sessions of the same sport in one duplicate cluster).
* Returns the positive impulses, one per session. Pure; exported for tests.
*/
export function canonicalWorkoutTrimps<
T extends WorkoutPickerRow & { id: string },
>(
rows: readonly T[],
priorityJson: unknown,
trimpOf: (row: T) => number,
): number[] {
// By id: the picker may hand back a copy of a row (it fills a missing heart
// rate from a WHOOP twin).
const picked = new Set(
pickCanonicalWorkoutRows(rows, priorityJson).map((row) => row.id),
);
const ladder = workoutSourceLadder(priorityJson);
const rankOf = (source: string) => {
const index = ladder.indexOf(source as (typeof ladder)[number]);
return index === -1 ? ladder.length : index;
};
const slots = new Map<
string,
{ picked: Map<string, number>; dropped: number }
>();
const keys = workoutDuplicateClusterKeys(rows);
for (const [index, row] of rows.entries()) {
const slot = slots.get(keys[index]!) ?? { picked: new Map(), dropped: 0 };
const trimp = trimpOf(row);
if (trimp > 0) {
if (picked.has(row.id)) {
slot.picked.set(row.source, (slot.picked.get(row.source) ?? 0) + trimp);
} else {
slot.dropped = Math.max(slot.dropped, trimp);
}
}
slots.set(keys[index]!, slot);
}
const out: number[] = [];
for (const slot of slots.values()) {
let best: { rank: number; trimp: number } | null = null;
for (const [source, trimp] of slot.picked) {
const rank = rankOf(source);
if (
best === null ||
rank < best.rank ||
(rank === best.rank && trimp > best.trimp)
) {
best = { rank, trimp };
}
}
const trimp = best !== null ? best.trimp : slot.dropped;
if (trimp > 0) out.push(trimp);
}
return out;
}
/**
* Build the Strain profile (age + sex) and resting HR from the user row +
* the latest resting-HR reading. `prisma` is the (worker) client.
@@ -50,8 +50,23 @@ vi.mock("@/lib/insights/status-provider", () => ({
vi.mock("@/lib/ai/coach/bytes-codec", () => ({
encryptToBytes: vi.fn(() => new Uint8Array([1, 2, 3])),
}));
vi.mock("@/lib/measurements/pick-canonical-workout-rows", () => ({
pickCanonicalWorkoutRows: vi.fn((rows) => rows),
vi.mock("@/lib/workouts/canonical-read", () => ({
readWorkoutsAroundRange: vi.fn(
async (
range: { from: Date | null; to: Date | null },
read: (bounds: { gte?: Date; lt?: Date }) => Promise<unknown[]>,
) => ({
rows: await read({
gte: range.from ?? undefined,
lt: range.to ?? undefined,
}),
range,
__windowed: true,
}),
),
canonicalWorkoutsInRange: vi.fn((windowed: { rows: unknown[] }) => [
...windowed.rows,
]),
}));
import {
@@ -636,8 +651,8 @@ describe("Activity Insight — locale", () => {
*/
describe("Activity Insight — own-history comparison", () => {
it("routes the history rows through the canonical picker", async () => {
const { pickCanonicalWorkoutRows } =
await import("@/lib/measurements/pick-canonical-workout-rows");
const { canonicalWorkoutsInRange } =
await import("@/lib/workouts/canonical-read");
vi.mocked(prisma.user.findUnique).mockResolvedValue({
dateOfBirth: null,
locale: "en",
@@ -650,20 +665,27 @@ describe("Activity Insight — own-history comparison", () => {
await runWorkoutInsightGenerate({ userId: USER, workoutId: WORKOUT }, NOW);
expect(pickCanonicalWorkoutRows).toHaveBeenCalledTimes(1);
expect(canonicalWorkoutsInRange).toHaveBeenCalledTimes(1);
// The user's own source ladder decides the winner, not a default.
expect(vi.mocked(pickCanonicalWorkoutRows).mock.calls[0][1]).toEqual({
expect(vi.mocked(canonicalWorkoutsInRange).mock.calls[0][1]).toEqual({
ladder: ["APPLE_HEALTH"],
});
});
it("excludes the workout being described from its own baseline", async () => {
it("excludes the workout being described, twins included, from its own baseline", async () => {
const { canonicalWorkoutsInRange } =
await import("@/lib/workouts/canonical-read");
await runWorkoutInsightGenerate({ userId: USER, workoutId: WORKOUT }, NOW);
const where = vi.mocked(prisma.workout.findMany).mock.calls[0][0]
?.where as { userId: string; id: { not: string } };
?.where as { userId: string; id?: unknown };
expect(where.userId).toBe(USER);
expect(where.id).toEqual({ not: WORKOUT });
// The row stays in the read so its whole duplicate cluster can be left
// out after the pick; dropping it in SQL left a twin behind.
expect(where.id).toBeUndefined();
expect(vi.mocked(canonicalWorkoutsInRange).mock.calls[0][2]).toEqual({
excludeClusterOf: WORKOUT,
});
});
});
+32 -29
View File
@@ -44,7 +44,10 @@ import {
WORKOUT_INSIGHT_PROMPT_VERSION,
} from "@/lib/ai/prompts/workout-insight";
import { prisma } from "@/lib/db";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
} from "@/lib/workouts/canonical-read";
import type { Locale } from "@/lib/i18n/config";
import { resolveJobLocale } from "@/lib/i18n/job-locale";
import { finalizeStatusSummary } from "@/lib/insights/status-shared";
@@ -357,37 +360,37 @@ export async function runWorkoutInsightGenerate(
whoopZoneDurations: parseWhoopZoneDurations(row.metadata),
});
// The current workout is excluded in SQL before canonicalization. The
// canonical picker then collapses cross-source twins before any median or
// prompt is built.
const historyRows = await prisma.workout.findMany({
where: {
userId,
sportType: row.sportType,
id: { not: row.id },
startedAt: {
gte: new Date(
row.startedAt.getTime() - OWN_HISTORY_LOOKBACK_DAYS * MS_PER_DAY,
),
lt: row.startedAt,
},
// The canonical sessions of the sport before this one, without this
// session's duplicate cluster: a twin a second source recorded a minute
// earlier is not the previous session.
const historyRows = await readWorkoutsAroundRange(
{
from: new Date(
row.startedAt.getTime() - OWN_HISTORY_LOOKBACK_DAYS * MS_PER_DAY,
),
to: row.startedAt,
},
orderBy: { startedAt: "desc" },
take: OWN_HISTORY_MAX_ROWS,
select: {
id: true,
source: true,
startedAt: true,
sportType: true,
durationSec: true,
avgHeartRate: true,
totalDistanceM: true,
totalEnergyKcal: true,
},
});
const history = pickCanonicalWorkoutRows(
(bounds) =>
prisma.workout.findMany({
where: { userId, sportType: row.sportType, startedAt: bounds },
orderBy: [{ startedAt: "desc" }, { id: "asc" }],
take: OWN_HISTORY_MAX_ROWS,
select: {
id: true,
source: true,
startedAt: true,
sportType: true,
durationSec: true,
avgHeartRate: true,
totalDistanceM: true,
totalEnergyKcal: true,
},
}),
);
const history = canonicalWorkoutsInRange(
historyRows,
profile?.sourcePriorityJson ?? null,
{ excludeClusterOf: row.id },
);
const evidence = buildWorkoutInsightEvidence({
@@ -1,6 +1,6 @@
import { describe, expect, it } from "vitest";
import { canonicalWorkoutTrimps } from "@/lib/insights/strain-score";
import { canonicalWorkoutTrimps } from "@/lib/workouts/canonical-read";
import {
pickCanonicalWorkoutRows,
@@ -133,8 +133,20 @@ function deduplicateWinningRows<T extends WorkoutPickerRow>(rows: T[]): T[] {
* within the window straddle a slot boundary, the clusters are exactly the
* old slots: every pair inside one slot is less than five minutes apart, and
* a cluster cannot reach past its slot without such a pair.
*
* Because a cluster is anchored on its first start, which cluster a workout
* lands in depends on the workouts before it. A reader that decides which row
* is canonical therefore reads through `readWorkoutsAroundRange` in
* `src/lib/workouts/canonical-read.ts`, which widens the read until the
* clusters of the requested range no longer depend on where the read began.
*/
const DUPLICATE_WINDOW_MS = 5 * 60 * 1000;
export const WORKOUT_DUPLICATE_WINDOW_MS = 5 * 60 * 1000;
const DUPLICATE_WINDOW_MS = WORKOUT_DUPLICATE_WINDOW_MS;
/** Rows of different groups never share a duplicate cluster. */
export function workoutClusterGroup(row: WorkoutPickerRow): string {
return `${row.userId ?? ""}:${row.sportType}`;
}
/**
* The duplicate cluster of every row, aligned with `rows`. Rows of different
@@ -151,7 +163,7 @@ export function workoutDuplicateClusterKeys(
): string[] {
const groups = new Map<string, number[]>();
rows.forEach((row, index) => {
const group = `${row.userId ?? ""}:${row.sportType}`;
const group = workoutClusterGroup(row);
const list = groups.get(group);
if (list) list.push(index);
else groups.set(group, [index]);
+1 -1
View File
@@ -336,7 +336,7 @@ export const workoutPaths: NonNullable<ZodOpenApiObject["paths"]> = {
tags: ["Measurements"],
summary: "List workouts (v1.4.32)",
description:
"Paginated workout list with cross-source canonical-row dedup. The picker reads the per-user source-priority ladder and buckets by `(startedAt 5 min slot, sportType)` — twin workouts (Apple Watch + Withings ScanWatch) collapse to a single canonical row per cluster. Field names mirror the iOS handoff contract: `distanceM`, `activeEnergyKcal`, `avgHr`, `maxHr`.",
"Paginated workout list with cross-source canonical-row dedup. The picker reads the per-user source-priority ladder and clusters workouts of one sport whose starts lie less than five minutes after the cluster's first start — twin workouts (Apple Watch + Withings ScanWatch) collapse to a single canonical row per cluster. A twin just outside `since` / `until` still decides which row inside the range is canonical, so the list agrees with `canonicalId` on the detail. Field names mirror the iOS handoff contract: `distanceM`, `activeEnergyKcal`, `avgHr`, `maxHr`.",
requestParams: {
query: z.object({
limit: z.coerce.number().int().min(1).max(200).optional(),
@@ -0,0 +1,240 @@
/**
* The windowed read gives every surface the canonical workouts a read of the
* whole history would give, whatever range it asked for.
*
* The store is an in-memory array behind the same bounded-read callback the
* Prisma readers pass, so the property covers the reader's reach-back loop and
* the range filter, not a copy of them.
*/
import fc from "fast-check";
import { describe, expect, it } from "vitest";
import { assertAsyncProperty } from "../../../../tests/property/fast-check";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalClusterOf,
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
startsInRange,
type WorkoutStartBounds,
type WorkoutStartRange,
} from "../canonical-read";
type Source = "APPLE_HEALTH" | "WITHINGS" | "WHOOP" | "MANUAL";
interface Row {
id: string;
startedAt: Date;
durationSec: number;
sportType: string;
source: Source;
avgHeartRate: number | null;
maxHeartRate: number | null;
metadata: null;
}
const MIN = 60_000;
const BASE = Date.UTC(2026, 9, 1, 9, 0, 0);
function memoryStore(rows: readonly Row[]) {
let reads = 0;
const read = async (bounds: WorkoutStartBounds) => {
reads += 1;
return rows
.filter(
(row) =>
(!bounds.gte || row.startedAt >= bounds.gte) &&
(!bounds.lt || row.startedAt < bounds.lt),
)
.sort(
(a, b) =>
a.startedAt.getTime() - b.startedAt.getTime() ||
a.id.localeCompare(b.id),
);
};
return { read, reads: () => reads };
}
function sortedAll(rows: readonly Row[]): Row[] {
return [...rows].sort(
(a, b) =>
a.startedAt.getTime() - b.startedAt.getTime() || a.id.localeCompare(b.id),
);
}
/** What a read of the whole history says is canonical inside the range. */
function oracle(rows: readonly Row[], range: WorkoutStartRange): string[] {
return pickCanonicalWorkoutRows(sortedAll(rows), null)
.filter((row) => startsInRange(range, row.startedAt))
.map((row) => row.id)
.sort();
}
/**
* Workouts packed tightly enough that chains of starts under five minutes
* apart are common: the case where the cluster of a row depends on rows far
* before it.
*/
const rowsArb = fc
.array(
fc.record({
gapMin: fc.integer({ min: 0, max: 9 }),
gapSec: fc.integer({ min: 0, max: 59 }),
sportType: fc.constantFrom("running", "running", "cycling"),
source: fc.constantFrom<Source>(
"APPLE_HEALTH",
"WITHINGS",
"WHOOP",
"MANUAL",
),
durationMin: fc.integer({ min: 10, max: 60 }),
hr: fc.option(fc.integer({ min: 90, max: 170 }), { nil: null }),
}),
{ minLength: 1, maxLength: 40 },
)
.map((specs) => {
let at = BASE;
return specs.map((spec, index): Row => {
at += spec.gapMin * MIN + spec.gapSec * 1000;
return {
id: `w${String(index).padStart(2, "0")}`,
startedAt: new Date(at),
durationSec: spec.durationMin * 60,
sportType: spec.sportType,
source: spec.source,
avgHeartRate: spec.hr,
maxHeartRate: null,
metadata: null,
};
});
});
const rangeArb = fc.record({
fromOffsetMin: fc.option(fc.integer({ min: -10, max: 400 }), { nil: null }),
lengthMin: fc.option(fc.integer({ min: 0, max: 200 }), { nil: null }),
inclusiveEnd: fc.boolean(),
});
function toRange(spec: {
fromOffsetMin: number | null;
lengthMin: number | null;
inclusiveEnd: boolean;
}): WorkoutStartRange {
const from =
spec.fromOffsetMin === null
? null
: new Date(BASE + spec.fromOffsetMin * MIN);
const to =
spec.lengthMin === null
? null
: new Date((from?.getTime() ?? BASE) + spec.lengthMin * MIN);
return { from, to, inclusiveEnd: spec.inclusiveEnd };
}
describe("readWorkoutsAroundRange", () => {
it("resolves the range's canonical workouts as a whole-history read does", async () => {
await assertAsyncProperty(
fc.asyncProperty(rowsArb, rangeArb, async (rows, rangeSpec) => {
const range = toRange(rangeSpec);
const store = memoryStore(rows);
const windowed = await readWorkoutsAroundRange(range, store.read);
const got = canonicalWorkoutsInRange(windowed, null)
.map((row) => row.id)
.sort();
expect(got).toEqual(oracle(rows, range));
}),
300,
);
});
it("gives one workout the same cluster whichever range found it", async () => {
await assertAsyncProperty(
fc.asyncProperty(rowsArb, fc.nat(), async (rows, pick) => {
const target = rows[pick % rows.length]!;
const store = memoryStore(rows);
const point = await readWorkoutsAroundRange(
{ from: target.startedAt, to: target.startedAt, inclusiveEnd: true },
store.read,
);
const whole = await readWorkoutsAroundRange(
{ from: null, to: null },
store.read,
);
const fromPoint = canonicalClusterOf(point, target.id, null);
const fromWhole = canonicalClusterOf(whole, target.id, null);
expect(fromPoint?.members.map((row) => row.id)).toEqual(
fromWhole?.members.map((row) => row.id),
);
expect(fromPoint?.canonical[0]?.id).toBe(fromWhole?.canonical[0]?.id);
}),
300,
);
});
it("depends on the history before the range, which the range alone does not show", async () => {
// Three runs four minutes apart. Read from 09:55 the 09:58 run anchors a
// cluster with 10:02, and 10:06 stands alone; read from 10:00 the 10:02
// run anchors one with 10:06 and hides it behind the Apple row. The
// windowed read reaches back to 09:58 and draws the clusters a
// whole-history read draws.
const at = (h: number, m: number) => new Date(Date.UTC(2026, 9, 1, h, m));
const rows: Row[] = [
["a", at(9, 58), "WITHINGS"],
["b", at(10, 2), "APPLE_HEALTH"],
["c", at(10, 6), "WITHINGS"],
].map(([id, startedAt, source]) => ({
id: id as string,
startedAt: startedAt as Date,
durationSec: 1800,
sportType: "running",
source: source as Source,
avgHeartRate: null,
maxHeartRate: null,
metadata: null,
}));
const range = { from: at(10, 0), to: at(11, 0) };
// The range read alone drops the 10:06 run as a twin of 10:02.
const naive = pickCanonicalWorkoutRows(
rows.filter((row) => startsInRange(range, row.startedAt)),
null,
).map((row) => row.id);
expect(naive).toEqual(["b"]);
const windowed = await readWorkoutsAroundRange(
range,
memoryStore(rows).read,
);
expect(
canonicalWorkoutsInRange(windowed, null).map((row) => row.id),
).toEqual(["b", "c"]);
expect(oracle(rows, range)).toEqual(["b", "c"]);
});
it("leaves a workout's twins out of a baseline along with the workout", async () => {
const at = (m: number) => new Date(BASE + m * MIN);
const rows: Row[] = [
{ id: "old", startedAt: at(-600) },
{ id: "apple", startedAt: at(0) },
{ id: "withings", startedAt: at(1) },
].map((row, index) => ({
...row,
durationSec: 1800,
sportType: "running",
source: index === 2 ? "WITHINGS" : "APPLE_HEALTH",
avgHeartRate: null,
maxHeartRate: null,
metadata: null,
}));
const windowed = await readWorkoutsAroundRange(
{ from: at(-1000), to: null },
memoryStore(rows).read,
);
expect(
canonicalWorkoutsInRange(windowed, null, {
excludeClusterOf: "withings",
}).map((row) => row.id),
).toEqual(["old"]);
});
});
+73
View File
@@ -0,0 +1,73 @@
/**
* The write-time duplicate pick of `POST /api/workouts/batch`.
*
* The rows a batch is about to insert are clustered together with the stored
* workouts around them, read through the same windowed reader the list uses,
* so a cluster boundary falls where every read will later draw it. Clustering
* the batch alone could merge a row into a cluster the list then splits (a
* stored 09:58 run anchors a cluster with an incoming 10:02, which leaves an
* incoming 10:06 a session of its own) and drop a workout the list would have
* shown.
*/
import type { MeasurementSource, Prisma } from "@/generated/prisma/client";
import { prisma } from "@/lib/db";
import {
incomingWorkoutsToKeep,
readWorkoutsAroundRange,
} from "./canonical-read";
/** The indexes of the batch rows to write; the rest are duplicates. */
export async function batchWorkoutIndicesToKeep(
userId: string,
prepared: ReadonlyArray<{
index: number;
row: Prisma.WorkoutCreateManyInput;
}>,
priorityJson: unknown,
): Promise<Set<number>> {
if (prepared.length === 0) return new Set();
const incoming = prepared.map((p) => ({
index: p.index,
userId: p.row.userId,
sportType: p.row.sportType,
startedAt: p.row.startedAt as Date,
durationSec: p.row.durationSec,
// Zod's `measurementSourceEnum.optional().default("MANUAL")` guarantees
// `source` is set post-parse, but Prisma's `WorkoutCreateManyInput` types
// it optional.
source: (p.row.source ?? "MANUAL") as MeasurementSource,
avgHeartRate: p.row.avgHeartRate ?? null,
maxHeartRate: p.row.maxHeartRate ?? null,
metadata: p.row.metadata ?? null,
}));
const starts = incoming.map((row) => row.startedAt.getTime());
const stored = await readWorkoutsAroundRange(
{
from: new Date(Math.min(...starts)),
to: new Date(Math.max(...starts)),
inclusiveEnd: true,
},
(bounds) =>
prisma.workout.findMany({
where: { userId, startedAt: bounds },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
userId: true,
source: true,
sportType: true,
startedAt: true,
durationSec: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
},
}),
);
return new Set(
incomingWorkoutsToKeep(stored, incoming, priorityJson).map(
(row) => row.index,
),
);
}
+395
View File
@@ -0,0 +1,395 @@
/**
* The read every surface that decides which workout is canonical goes through.
*
* `pickCanonicalWorkoutRows` clusters workouts of one sport by start time,
* anchored on each cluster's first start (see `workoutDuplicateClusterKeys`).
* That makes the cluster of a workout depend on the workouts around it: on
* the twin a few minutes after it, which may lie past the end of the range a
* surface asked for, and on the run of workouts before it, which decides where
* the anchors fall. A surface that handed the picker only the rows inside its
* own range got an answer that depended on the range: the list and the detail
* of one workout disagreed about which row was canonical, and 09:58, 10:02 and
* 10:06 were grouped one way by a read starting at 09:55 and another way by a
* read starting at 10:00.
*
* `readWorkoutsAroundRange` reads the range plus one duplicate window after its
* end, and before its start as far as it takes to reach, for every sport with a
* row in the range, a workout that starts a full window after the one before
* it. Such a workout is an anchor whatever comes before it, so from there on
* the clusters are the ones a read of the whole history would produce. The
* surfaces then pick over everything read and keep the rows that start inside
* their range: a workout is canonical, or not, the same way on every surface.
*
* A read callback may still carry a row cap as a runaway guard. A capped read
* that actually hits its cap has lost rows at one end and is outside this
* guarantee; the caps sit far above any real account's volume.
*/
import {
pickCanonicalWorkoutRows,
WORKOUT_DUPLICATE_WINDOW_MS,
workoutClusterGroup,
workoutDuplicateClusterKeys,
workoutSourceLadder,
type WorkoutPickerRow,
} from "@/lib/measurements/pick-canonical-workout-rows";
/**
* The columns a read needs for the canonical pick and nothing more: the
* session identity (`sportType`, `startedAt`, `durationSec`), the source the
* ladder ranks, and the id the windowed read orders by.
*/
export const WORKOUT_PICK_SELECT = {
id: true,
source: true,
sportType: true,
startedAt: true,
durationSec: true,
} as const;
/** A row the windowed read can order deterministically. */
export type CanonicalReadRow = WorkoutPickerRow & { id: string };
/**
* The range of starts a surface asks about. `null` leaves that side open.
* `inclusiveEnd` makes `to` part of the range (`lte`), as the list's `until`
* and the doctor report's end are.
*/
export interface WorkoutStartRange {
from: Date | null;
to: Date | null;
inclusiveEnd?: boolean;
}
/** The `startedAt` bounds one read callback must apply. */
export interface WorkoutStartBounds {
gte?: Date;
lt?: Date;
}
/**
* Reads the caller's workouts whose start lies inside `bounds`, with every
* other filter (user, sport) the caller applies. Called more than once when
* the read has to reach further back.
*/
export type WorkoutBoundedRead<T extends CanonicalReadRow> = (
bounds: WorkoutStartBounds,
) => Promise<T[]>;
/**
* Rows read for a range, in `startedAt ASC, id ASC` order, together with the
* range they were read for. Only `readWorkoutsAroundRange` builds one, so a
* picker call below cannot be handed rows that were read without the margin.
*/
export interface WindowedWorkouts<T extends CanonicalReadRow> {
readonly rows: readonly T[];
readonly range: WorkoutStartRange;
readonly __windowed: true;
}
/**
* How many times the read reaches further back before it stops. Each step
* doubles the reach, so the last one covers more than a week of workouts
* every one of which started less than five minutes after the previous one.
*/
const MAX_BACKWARD_STEPS = 12;
function compareRows(a: CanonicalReadRow, b: CanonicalReadRow): number {
const delta = a.startedAt.getTime() - b.startedAt.getTime();
if (delta !== 0) return delta;
return a.id < b.id ? -1 : a.id > b.id ? 1 : 0;
}
/**
* The earliest start that is not yet known to be an anchor, among the sports
* that have a row at or after `from`. A sport's earliest row read is an anchor
* once the read reaches a full window before it (nothing of that sport starts
* inside that window, or the read would hold it). `null` when every sport is
* settled.
*/
function earliestUnsettledStart(
rows: readonly CanonicalReadRow[],
from: Date,
lowerMs: number,
): number | null {
const earliest = new Map<string, number>();
const relevant = new Set<string>();
for (const row of rows) {
const group = workoutClusterGroup(row);
const start = row.startedAt.getTime();
const known = earliest.get(group);
if (known === undefined || start < known) earliest.set(group, start);
if (start >= from.getTime()) relevant.add(group);
}
let unsettled: number | null = null;
for (const group of relevant) {
const start = earliest.get(group)!;
if (start - WORKOUT_DUPLICATE_WINDOW_MS < lowerMs) {
unsettled = unsettled === null ? start : Math.min(unsettled, start);
}
}
return unsettled;
}
export async function readWorkoutsAroundRange<T extends CanonicalReadRow>(
range: WorkoutStartRange,
read: WorkoutBoundedRead<T>,
): Promise<WindowedWorkouts<T>> {
const window = WORKOUT_DUPLICATE_WINDOW_MS;
// A cluster is never wider than one window from its anchor, and an anchor
// of the range starts no later than the range's end.
const upper = range.to ? new Date(range.to.getTime() + window) : undefined;
const from = range.from;
let lowerMs = from ? from.getTime() - window : null;
const rows: T[] = [
...(await read({
gte: lowerMs === null ? undefined : new Date(lowerMs),
lt: upper,
})),
];
if (from && lowerMs !== null) {
for (let step = 0; step < MAX_BACKWARD_STEPS; step += 1) {
const unsettled = earliestUnsettledStart(rows, from, lowerMs);
if (unsettled === null) break;
const nextMs: number = Math.min(
unsettled - window,
lowerMs - window * 2 ** step,
);
const earlier = await read({
gte: new Date(nextMs),
lt: new Date(lowerMs),
});
lowerMs = nextMs;
// Nothing between the new and the old bound: every unsettled sport now
// has a full window of nothing before its earliest row.
if (earlier.length === 0) break;
rows.push(...earlier);
}
}
rows.sort(compareRows);
return { rows, range, __windowed: true };
}
/** Whether a start lies inside the range a surface asked about. */
export function startsInRange(range: WorkoutStartRange, at: Date): boolean {
const t = at.getTime();
if (range.from && t < range.from.getTime()) return false;
if (range.to) {
const end = range.to.getTime();
if (range.inclusiveEnd ? t > end : t >= end) return false;
}
return true;
}
/** The rows read that start inside the requested range. */
export function rowsInRange<T extends CanonicalReadRow>(
windowed: WindowedWorkouts<T>,
): T[] {
return windowed.rows.filter((row) =>
startsInRange(windowed.range, row.startedAt),
);
}
function clusterKeyById<T extends CanonicalReadRow>(
windowed: WindowedWorkouts<T>,
): Map<string, string> {
const keys = workoutDuplicateClusterKeys(windowed.rows);
return new Map(windowed.rows.map((row, index) => [row.id, keys[index]!]));
}
/**
* The canonical workouts that start inside the range, oldest first.
*
* `excludeClusterOf` leaves out the cluster of one workout: a baseline that
* compares a session against the others of its sport must drop the session's
* twins along with the session itself. Removing the row from the read instead
* left its twin behind as a second session, and moved the cluster anchors of
* whatever followed.
*/
export function canonicalWorkoutsInRange<T extends CanonicalReadRow>(
windowed: WindowedWorkouts<T>,
priorityJson: unknown,
options: { excludeClusterOf?: string } = {},
): T[] {
const picked = pickCanonicalWorkoutRows(windowed.rows, priorityJson).filter(
(row) => startsInRange(windowed.range, row.startedAt),
);
if (!options.excludeClusterOf) return picked;
const keyById = clusterKeyById(windowed);
const excluded = keyById.get(options.excludeClusterOf);
if (excluded === undefined) return picked;
return picked.filter((row) => keyById.get(row.id) !== excluded);
}
/**
* The duplicate cluster of one workout: its rows in start order, and the
* canonical ones among them, nearest to the workout's own start first. `null`
* when the workout is not among the rows read.
*/
export function canonicalClusterOf<T extends CanonicalReadRow>(
windowed: WindowedWorkouts<T>,
workoutId: string,
priorityJson: unknown,
): { members: T[]; canonical: T[] } | null {
const self = windowed.rows.find((row) => row.id === workoutId);
if (!self) return null;
const keyById = clusterKeyById(windowed);
const key = keyById.get(workoutId);
const members = windowed.rows.filter((row) => keyById.get(row.id) === key);
const at = self.startedAt.getTime();
const canonical = pickCanonicalWorkoutRows(members, priorityJson).sort(
(a, b) =>
Number(b.id === workoutId) - Number(a.id === workoutId) ||
Math.abs(a.startedAt.getTime() - at) -
Math.abs(b.startedAt.getTime() - at) ||
compareRows(a, b),
);
return { members, canonical };
}
/**
* The write-time pick of the batch route. The incoming rows are clustered
* together with the stored workouts around them, so a cluster boundary falls
* where every read will later draw it. An incoming row is dropped only when
* another incoming row of its cluster is the one kept: a row that loses to a
* stored workout is still written (the reads hide it), as before.
*
* Returns the incoming rows to keep, in their input order.
*/
export function incomingWorkoutsToKeep<
S extends CanonicalReadRow,
I extends WorkoutPickerRow,
>(
stored: WindowedWorkouts<S>,
incoming: readonly I[],
priorityJson: unknown,
): I[] {
type Tagged = WorkoutPickerRow & { tag: number };
// Stored rows first on an equal start, which is where a read that orders by
// start finds them relative to rows written after them.
const union: Tagged[] = [
...stored.rows.map((row) => ({ ...row, tag: -1 })),
...incoming.map((row, index) => ({ ...row, tag: index })),
].sort(
(a, b) =>
a.startedAt.getTime() - b.startedAt.getTime() ||
Number(a.tag >= 0) - Number(b.tag >= 0) ||
a.tag - b.tag,
);
const keys = workoutDuplicateClusterKeys(union);
const keyByTag = new Map<number, string>();
union.forEach((row, index) => {
if (row.tag >= 0) keyByTag.set(row.tag, keys[index]!);
});
const keptTags = new Set(
pickCanonicalWorkoutRows(union, priorityJson)
.map((row) => row.tag)
.filter((tag) => tag >= 0),
);
const clustersWithIncomingWinner = new Set(
[...keptTags].map((tag) => keyByTag.get(tag)!),
);
return incoming.filter(
(_, index) =>
keptTags.has(index) ||
!clustersWithIncomingWinner.has(keyByTag.get(index)!),
);
}
/**
* One training impulse per workout SESSION, not per stored row.
*
* The same session can be stored more than once: the watch writes it to Apple
* Health and a second service (Strava, WHOOP, a ring) syncs it again, or
* HealthKit re-sends it under a new identifier. The workout list collapses
* those through `pickCanonicalWorkoutRows`; summing every row here counted the
* session twice in the day's TRIMP while the list showed it once.
*
* The picker chooses the row by source order, which says nothing about heart
* rate: the winning row can be one without a series while its dropped twin
* carries one. So the impulse of a slot is the sum over its picked rows, and
* only when that is zero the largest impulse among the slot's dropped rows.
*
* The picker keeps every row of a slot none of whose sources is on the
* ladder, so the picked rows of one slot can come from several sources. Only
* one source's rows count: the best-ranked on the ladder, a source off the
* ladder ranking below every listed one, and among sources off the ladder
* the one with the largest impulse. Rows of that one source still add up
* (two distinct sessions of the same sport in one duplicate cluster).
*
* `belongs` keeps the sessions of the scored range: the rows come from a
* windowed read that reaches past the day on both sides, and a session counts
* on the day its earliest picked row starts, the day the list shows it on.
* Returns the positive impulses, one per session. Pure; the strain
* engine's day TRIMP sums it, and the tests call it directly.
*/
export function canonicalWorkoutTrimps<
T extends WorkoutPickerRow & { id: string },
>(
rows: readonly T[],
priorityJson: unknown,
trimpOf: (row: T) => number,
belongs: (startedAt: Date) => boolean = () => true,
): number[] {
// By id: the picker may hand back a copy of a row (it fills a missing heart
// rate from a WHOOP twin).
const picked = new Set(
pickCanonicalWorkoutRows(rows, priorityJson).map((row) => row.id),
);
const ladder = workoutSourceLadder(priorityJson);
const rankOf = (source: string) => {
const index = ladder.indexOf(source as (typeof ladder)[number]);
return index === -1 ? ladder.length : index;
};
const slots = new Map<
string,
{
picked: Map<string, number>;
dropped: number;
firstPickedAt: Date | null;
}
>();
const keys = workoutDuplicateClusterKeys(rows);
for (const [index, row] of rows.entries()) {
const slot = slots.get(keys[index]!) ?? {
picked: new Map(),
dropped: 0,
firstPickedAt: null,
};
if (
picked.has(row.id) &&
(slot.firstPickedAt === null || row.startedAt < slot.firstPickedAt)
) {
slot.firstPickedAt = row.startedAt;
}
const trimp = trimpOf(row);
if (trimp > 0) {
if (picked.has(row.id)) {
slot.picked.set(row.source, (slot.picked.get(row.source) ?? 0) + trimp);
} else {
slot.dropped = Math.max(slot.dropped, trimp);
}
}
slots.set(keys[index]!, slot);
}
const out: number[] = [];
for (const slot of slots.values()) {
if (slot.firstPickedAt === null || !belongs(slot.firstPickedAt)) continue;
let best: { rank: number; trimp: number } | null = null;
for (const [source, trimp] of slot.picked) {
const rank = rankOf(source);
if (
best === null ||
rank < best.rank ||
(rank === best.rank && trimp > best.trimp)
) {
best = { rank, trimp };
}
}
const trimp = best !== null ? best.trimp : slot.dropped;
if (trimp > 0) out.push(trimp);
}
return out;
}
+51 -46
View File
@@ -18,7 +18,11 @@ import { z } from "zod/v4";
import { prisma } from "@/lib/db";
import { annotate } from "@/lib/logging/context";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
rowsInRange,
} from "@/lib/workouts/canonical-read";
import { cached, caches } from "@/lib/cache/server-cache";
import { workoutSportTypeEnum } from "@/lib/validations/workout";
import { isoDateTime } from "@/lib/validations/datetime";
@@ -131,62 +135,63 @@ export async function buildWorkoutsProjection(
): Promise<WorkoutsProjection> {
const { since, until, sportType } = params;
const where: Record<string, unknown> = { userId };
if (since || until) {
const range: Record<string, Date> = {};
if (since) {
const d = new Date(since);
if (!Number.isNaN(d.getTime())) range.gte = d;
}
if (until) {
const d = new Date(until);
if (!Number.isNaN(d.getTime())) range.lte = d;
}
if (Object.keys(range).length > 0) {
where.startedAt = range;
}
}
if (sportType) {
where.sportType = sportType;
}
const parse = (value: string | null): Date | null => {
if (!value) return null;
const d = new Date(value);
return Number.isNaN(d.getTime()) ? null : d;
};
const range = {
from: parse(since),
to: parse(until),
inclusiveEnd: true,
};
// Independent reads — the sourcePriorityJson lookup doesn't depend on the
// row read, so run them concurrently rather than paying two sequential
// round trips.
const [rows, userRow] = await Promise.all([
prisma.workout.findMany({
where,
orderBy: [{ startedAt: "desc" }, { id: "asc" }],
select: {
id: true,
source: true,
externalId: true,
sportType: true,
startedAt: true,
endedAt: true,
durationSec: true,
totalDistanceM: true,
totalEnergyKcal: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
createdAt: true,
route: { select: { id: true } },
samples: { select: { sampleCount: true } },
},
}),
// round trips. The rows come through the windowed reader: a twin just past
// `until`, or a cluster that began just before `since`, decides which row
// of the range is canonical exactly as it does for the detail route.
const [around, userRow] = await Promise.all([
readWorkoutsAroundRange(range, (bounds) =>
prisma.workout.findMany({
where: {
userId,
startedAt: bounds,
...(sportType ? { sportType } : {}),
},
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
select: {
id: true,
source: true,
externalId: true,
sportType: true,
startedAt: true,
endedAt: true,
durationSec: true,
totalDistanceM: true,
totalEnergyKcal: true,
avgHeartRate: true,
maxHeartRate: true,
metadata: true,
createdAt: true,
route: { select: { id: true } },
samples: { select: { sampleCount: true } },
},
}),
),
prisma.user.findUnique({
where: { id: userId },
select: { sourcePriorityJson: true },
}),
]);
const canonical = pickCanonicalWorkoutRows(
rows,
const canonical: CanonicalWorkoutRow[] = canonicalWorkoutsInRange(
around,
userRow?.sourcePriorityJson ?? null,
);
const rawCount = rowsInRange(around).length;
// `pickCanonicalWorkoutRows` re-sorts its output to `startedAt ASC` (its
// The canonical pick comes back in `startedAt ASC` order (its
// documented contract for the rollup callers). This surface is a
// most-recent-first list, so the DB `orderBy startedAt desc` must be
// re-established AFTER the pick — otherwise `offset`/`limit` slice the
@@ -197,7 +202,7 @@ export async function buildWorkoutsProjection(
return a.id.localeCompare(b.id);
});
return { canonical, rawCount: rows.length };
return { canonical, rawCount };
}
export function sliceWorkoutsProjection(
+29 -22
View File
@@ -18,7 +18,10 @@ import {
type CoachHistoryReach,
} from "@/lib/ai/coach/history-reach";
import { prisma } from "@/lib/db";
import { pickCanonicalWorkoutRows } from "@/lib/measurements/pick-canonical-workout-rows";
import {
canonicalWorkoutsInRange,
readWorkoutsAroundRange,
} from "@/lib/workouts/canonical-read";
/** Own-history lookback for the sport-average comparison line. */
const SPORT_CONTEXT_LOOKBACK_DAYS = 180;
@@ -47,28 +50,32 @@ export async function buildSportContext(
const since = new Date(
Date.now() - SPORT_CONTEXT_LOOKBACK_DAYS * 24 * 60 * 60 * 1000,
);
const rows = await prisma.workout.findMany({
// `userId` stays in the where clause — the universal tenancy narrow. The
// selected workout is not part of the baseline it is compared against.
where: {
userId,
sportType,
startedAt: { gte: since },
...(excludeWorkoutId ? { id: { not: excludeWorkoutId } } : {}),
},
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
take: SPORT_CONTEXT_MAX_ROWS,
select: {
id: true,
source: true,
startedAt: true,
sportType: true,
durationSec: true,
totalDistanceM: true,
avgHeartRate: true,
},
// `userId` stays in the where clause — the universal tenancy narrow. The
// selected workout is not part of the baseline it is compared against, and
// neither are its twins: the selected row stays in the read so its whole
// duplicate cluster can be left out after the pick. Dropping only the row in
// SQL left a twin from a second source behind as a session of its own.
const rows = await readWorkoutsAroundRange(
{ from: since, to: null },
(bounds) =>
prisma.workout.findMany({
where: { userId, sportType, startedAt: bounds },
orderBy: [{ startedAt: "asc" }, { id: "asc" }],
take: SPORT_CONTEXT_MAX_ROWS,
select: {
id: true,
source: true,
startedAt: true,
sportType: true,
durationSec: true,
totalDistanceM: true,
avgHeartRate: true,
},
}),
);
const canonical = canonicalWorkoutsInRange(rows, sourcePriorityJson, {
excludeClusterOf: excludeWorkoutId,
});
const canonical = pickCanonicalWorkoutRows(rows, sourcePriorityJson);
if (canonical.length === 0) return null;
let durationSum = 0;
@@ -0,0 +1,215 @@
/**
* The workout list and the workout detail resolve a duplicate cluster the same
* way, whatever range either of them read.
*
* - Two copies of one run thirty seconds apart on either side of a
* five-minute clock boundary: the list shows the Apple Health row, and the
* detail of the Withings copy names that row as `canonicalId`. The detail
* used to read only the clock slot of the requested row, found no twin,
* and declared the hidden copy canonical.
* - Runs at 09:58 (Withings), 10:02 (Apple Health) and 10:06 (Withings): a
* whole-history read pairs 09:58 with 10:02 and leaves 10:06 alone. A list
* filtered from 10:00 shows 10:02 and 10:06, the detail of every row agrees
* with that, and the session before 10:06 is 10:02.
* - The batch write clusters what it is about to insert together with the
* stored workouts around it: with 09:58 stored, a batch of 10:02 (Apple
* Health) and 10:06 (entered by hand) keeps both, where clustering the
* batch alone dropped 10:06 as a twin of 10:02 although every read shows
* it as a session of its own.
*/
import { beforeEach, describe, expect, it } from "vitest";
import type { User } from "@/generated/prisma/client";
import { GET as getDetail } from "@/app/api/workouts/[id]/route";
import { GET as getList } from "@/app/api/workouts/route";
import { POST as postBatch } from "@/app/api/workouts/batch/route";
import { callRoute, type RouteAuth } from "../../kit/call-route";
import { makeSession, makeUser, makeWorkout } from "../../kit/factories";
import { getPrismaClient, truncateAllTables } from "../setup";
const prisma = getPrismaClient();
beforeEach(() => truncateAllTables(prisma));
const DAY_MS = 24 * 60 * 60 * 1000;
/** A day ten days back, so the list's default window always holds it. */
const DAY = new Date(Date.now() - 10 * DAY_MS).toISOString().slice(0, 10);
/** An instant on that day, `HH:MM:SS` in UTC. */
const at = (time: string) => `${DAY}T${time}.000Z`;
interface ListBody {
workouts: Array<{ id: string }>;
}
interface DetailBody {
id: string;
canonicalId: string;
previousWorkoutId: string | null;
}
async function run(
user: Pick<User, "id">,
startedAt: string,
source: "APPLE_HEALTH" | "WITHINGS",
) {
return makeWorkout(user, {
startedAt: new Date(startedAt),
endedAt: new Date(new Date(startedAt).getTime() + 30 * 60 * 1000),
durationSec: 30 * 60,
source,
externalId: `${source}-${startedAt}`,
});
}
async function list(auth: RouteAuth, query: Record<string, string> = {}) {
const res = await callRoute<ListBody>(getList, {
method: "GET",
route: "/api/workouts",
query,
auth,
});
expect(res.status).toBe(200);
return res.data!.workouts.map((w) => w.id);
}
async function detail(auth: RouteAuth, id: string) {
const res = await callRoute<DetailBody>(getDetail, {
method: "GET",
route: "/api/workouts/{id}",
params: { id },
query: { compact: "1" },
auth,
});
expect(res.status).toBe(200);
return res.data!;
}
describe("list and detail agree on the canonical workout", () => {
it("resolves a twin across a clock-slot boundary to the row the list shows", async () => {
const user = await makeUser();
const auth = await makeSession(user);
const apple = await run(user, at("08:04:50"), "APPLE_HEALTH");
const withings = await run(user, at("08:05:20"), "WITHINGS");
expect(await list(auth)).toEqual([apple.id]);
expect((await detail(auth, apple.id)).canonicalId).toBe(apple.id);
// The deep link to the hidden copy lands on the row the list shows.
expect((await detail(auth, withings.id)).canonicalId).toBe(apple.id);
});
it("draws the clusters of a filtered range as a read of the whole history does", async () => {
const user = await makeUser();
const auth = await makeSession(user);
const early = await run(user, at("09:58:00"), "WITHINGS");
const middle = await run(user, at("10:02:00"), "APPLE_HEALTH");
const late = await run(user, at("10:06:00"), "WITHINGS");
expect(await list(auth)).toEqual([late.id, middle.id]);
// Filtered from 10:00 the list no longer holds the 09:58 run, and still
// draws the clusters the whole history draws: 10:06 is a session of its
// own, not a twin of 10:02.
expect(await list(auth, { since: at("10:00:00") })).toEqual([
late.id,
middle.id,
]);
// Up to 10:04 the list must still hide 10:02's twin from 09:58 and show
// 10:02, though the read stops before 10:06.
expect(
await list(auth, {
since: at("09:59:00"),
until: at("10:04:00"),
}),
).toEqual([middle.id]);
expect((await detail(auth, early.id)).canonicalId).toBe(middle.id);
expect((await detail(auth, middle.id)).canonicalId).toBe(middle.id);
const lateDetail = await detail(auth, late.id);
expect(lateDetail.canonicalId).toBe(late.id);
expect(lateDetail.previousWorkoutId).toBe(middle.id);
// The session before the 10:02 run is none: 09:58 is its own twin.
expect((await detail(auth, middle.id)).previousWorkoutId).toBeNull();
});
});
describe("the batch write draws clusters where the reads draw them", () => {
it("keeps a workout the reads show as a session of its own", async () => {
const user = await makeUser();
const auth = await makeSession(user);
await run(user, at("09:58:00"), "WITHINGS");
const res = await callRoute<{
entries: Array<{ index: number; status: string }>;
}>(postBatch, {
method: "POST",
route: "/api/workouts/batch",
body: {
workouts: [
{
sportType: "running",
startedAt: at("10:02:00"),
endedAt: at("10:32:00"),
source: "APPLE_HEALTH",
externalId: "apple-1002",
},
{
sportType: "running",
startedAt: at("10:06:00"),
endedAt: at("10:36:00"),
source: "MANUAL",
externalId: "manual-1006",
},
],
},
auth,
});
expect(res.status).toBe(200);
expect(res.data!.entries.map((e) => e.status)).toEqual([
"inserted",
"inserted",
]);
const stored = await prisma.workout.findMany({
where: { userId: user.id },
orderBy: { startedAt: "asc" },
select: { id: true, externalId: true },
});
expect(await list(auth)).toEqual([stored[2]!.id, stored[1]!.id]);
});
it("still drops the twin a batch carries of its own winner", async () => {
const user = await makeUser();
const auth = await makeSession(user);
const res = await callRoute<{
entries: Array<{ index: number; status: string }>;
}>(postBatch, {
method: "POST",
route: "/api/workouts/batch",
body: {
workouts: [
{
sportType: "running",
startedAt: at("08:04:50"),
endedAt: at("08:34:50"),
source: "APPLE_HEALTH",
externalId: "apple-0804",
},
{
sportType: "running",
startedAt: at("08:05:20"),
endedAt: at("08:35:20"),
source: "MANUAL",
externalId: "manual-0805",
},
],
},
auth,
});
expect(res.status).toBe(200);
expect(res.data!.entries.map((e) => e.status)).toEqual([
"inserted",
"duplicate",
]);
});
});