diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts index 2f039b49f..b513065e9 100644 --- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts @@ -35,14 +35,26 @@ function row( return { id, type, value, measuredAt: new Date(iso), externalId, unit }; } -function buildPrismaMock(rowsByType: Record) { - const upsert = vi.fn().mockResolvedValue({}); +function buildPrismaMock( + rowsByType: Record, + // When set, the canonical-row lookup inside `writeDay` resolves to this + // existing row id — simulating a daily row already sitting on the + // canonical local-noon instant (the P2002 coexistence case). + existingCanonicalId: string | null = null, +) { + const create = vi.fn().mockResolvedValue({ id: "minted-daily" }); + const update = vi.fn().mockResolvedValue({}); + const findFirst = vi + .fn() + .mockResolvedValue( + existingCanonicalId ? { id: existingCanonicalId } : null, + ); const updateMany = vi.fn().mockResolvedValue({ count: 0 }); const findManyMeasurement = vi.fn( async (args: { where: { type: string } }) => rowsByType[args.where.type] ?? [], ); - const tx = { measurement: { upsert, updateMany } }; + const tx = { measurement: { create, update, findFirst, updateMany } }; return { mock: { user: { @@ -55,7 +67,9 @@ function buildPrismaMock(rowsByType: Record) { cb(tx), ), } as unknown as PrismaClient, - upsert, + create, + update, + findFirst, updateMany, findManyMeasurement, }; @@ -119,13 +133,13 @@ describe("runDenseIntradayRetention — retention bound", () => { }); describe("runDenseIntradayRetention — fold flow", () => { - it("upserts the day MEAN and soft-deletes the out-of-window rows", async () => { + it("creates the day MEAN and soft-deletes the out-of-window rows when no canonical row exists", async () => { // Two HRV samples on an out-of-window day. const hrvRows = [ row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"), row("b", 60, "2026-05-01T09:00:00.000Z", "HEART_RATE_VARIABILITY"), ]; - const { mock, upsert, updateMany } = buildPrismaMock({ + const { mock, create, update, updateMany } = buildPrismaMock({ HEART_RATE_VARIABILITY: hrvRows, }); @@ -134,13 +148,15 @@ describe("runDenseIntradayRetention — fold flow", () => { log: () => {}, }); - const upsertArg = upsert.mock.calls[0]?.[0] as { - create: { value: number; source: string; type: string }; + // No pre-existing canonical row → create, not update. + expect(update).not.toHaveBeenCalled(); + const createArg = create.mock.calls[0]?.[0] as { + data: { value: number; source: string; type: string }; }; // value is the MEAN (50), not the sum (100). - expect(upsertArg.create.value).toBeCloseTo(50, 6); - expect(upsertArg.create.source).toBe("APPLE_HEALTH"); - expect(upsertArg.create.type).toBe("HEART_RATE_VARIABILITY"); + expect(createArg.data.value).toBeCloseTo(50, 6); + expect(createArg.data.source).toBe("APPLE_HEALTH"); + expect(createArg.data.type).toBe("HEART_RATE_VARIABILITY"); // soft-delete, never hard delete. const updArg = updateMany.mock.calls[0]?.[0] as { @@ -151,8 +167,41 @@ describe("runDenseIntradayRetention — fold flow", () => { expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(1); }); + it("adopts an existing canonical row in place instead of colliding (P2002 coexistence)", async () => { + const hrvRows = [ + row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"), + row("b", 60, "2026-05-01T09:00:00.000Z", "HEART_RATE_VARIABILITY"), + ]; + // A daily row already sits on the canonical local-noon instant. + const { mock, create, update, updateMany } = buildPrismaMock( + { HEART_RATE_VARIABILITY: hrvRows }, + "existing-canonical-row", + ); + + await runDenseIntradayRetention(mock, { retentionDays: 0, log: () => {} }); + + // The fold adopts the existing row in place — no create (which would + // collide on the measured_at unique index), an update with the stats + // externalId + refreshed mean. + expect(create).not.toHaveBeenCalled(); + const updateArg = update.mock.calls[0]?.[0] as { + where: { id: string }; + data: { value: number; externalId: string; deletedAt: null }; + }; + expect(updateArg.where.id).toBe("existing-canonical-row"); + expect(updateArg.data.value).toBeCloseTo(50, 6); + expect(updateArg.data.externalId.startsWith("stats:")).toBe(true); + expect(updateArg.data.deletedAt).toBeNull(); + + // The soft-delete excludes the adopted canonical row id. + const updManyArg = updateMany.mock.calls[0]?.[0] as { + where: { id: { not: string } }; + }; + expect(updManyArg.where.id.not).toBe("existing-canonical-row"); + }); + it("does not write on a dry-run", async () => { - const { mock, upsert } = buildPrismaMock({ + const { mock, create, update } = buildPrismaMock({ HEART_RATE_VARIABILITY: [ row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"), ], @@ -162,7 +211,8 @@ describe("runDenseIntradayRetention — fold flow", () => { dryRun: true, log: () => {}, }); - expect(upsert).not.toHaveBeenCalled(); + expect(create).not.toHaveBeenCalled(); + expect(update).not.toHaveBeenCalled(); expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled(); }); }); diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index da61b9f6b..070ca8fdd 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -183,41 +183,81 @@ export async function runDenseIntradayRetention( const unit = dayRows[0]?.unit ?? "unknown"; let removed = 0; await pc.$transaction(async (tx) => { - // Mint / refresh the canonical daily-mean row first. The unique - // index (userId, type, source, externalId) makes the upsert - // idempotent across re-runs. Built field-by-field (no spread) per - // the no-mass-assignment convention. - await tx.measurement.upsert({ + // Fold the day to one canonical daily-mean row at local-noon. + // + // Unlike `consolidate-daily-mean` (whose types never share the + // canonical local-noon instant with another path), the dense-tier + // types HRV / PULSE are dense enough that a per-sample row can land + // exactly on the canonical timestamp, and a previously-minted daily + // row can already occupy it. An externalId-keyed upsert misses the + // lookup (the colliding row carries a different externalId) and the + // INSERT then violates the OTHER unique index + // `(userId, type, measuredAt, source, sleepStage)` with P2002. + // + // The canonical daily row is identified by BOTH unique keys at once: + // its measuredAt is always `canonicalTimestamp` (local-noon) and its + // externalId is always the `stats:` id, so the two indexes point at + // the same logical row. Resolve the row by the measured_at composite + // — the one the collision is on (`sleepStage` is NULL for these + // continuous types, matched via the NULLS-NOT-DISTINCT index) — and + // either adopt it in place or create it. Built field-by-field (no + // spread) per the no-mass-assignment convention. + const existing = await tx.measurement.findFirst({ where: { - userId_type_source_externalId: { - userId, - type, - source: "APPLE_HEALTH", - externalId, - }, - }, - create: { userId, type, - value: reducedValue, - unit, source: "APPLE_HEALTH", measuredAt: canonicalTimestamp, - externalId, - }, - update: { - value: reducedValue, - measuredAt: canonicalTimestamp, - deletedAt: null, + sleepStage: null, }, + select: { id: true }, }); + let canonicalRowId: string; + if (existing) { + // Adopt the row already sitting on the canonical instant: refresh + // its value, stamp the `stats:` externalId, and un-tombstone it. + // This coexists with a pre-existing daily row (a prior fold, a + // manual daily entry, or a per-sample row that happened at local + // noon) instead of colliding with it. + await tx.measurement.update({ + where: { id: existing.id }, + data: { + value: reducedValue, + unit, + externalId, + deletedAt: null, + }, + }); + canonicalRowId = existing.id; + } else { + const created = await tx.measurement.create({ + data: { + userId, + type, + value: reducedValue, + unit, + source: "APPLE_HEALTH", + measuredAt: canonicalTimestamp, + externalId, + }, + select: { id: true }, + }); + canonicalRowId = created.id; + } + // Soft-delete the out-of-window per-sample rows in the same // transaction — tombstone, never hard-delete; they remain as an // audit trail and drop off the live read + this pass's re-run - // discovery. + // discovery. EXCLUDE the canonical row itself: a per-sample row that + // happened to fall on the canonical local-noon instant is the row we + // just adopted as the daily mean, so tombstoning it would erase the + // fold. const del = await tx.measurement.updateMany({ - where: { id: { in: sourceRowIds }, deletedAt: null }, + where: { + id: { in: sourceRowIds, not: canonicalRowId }, + deletedAt: null, + }, data: { deletedAt: new Date() }, }); removed = del.count; diff --git a/tests/integration/dense-intraday-retention.test.ts b/tests/integration/dense-intraday-retention.test.ts new file mode 100644 index 000000000..1301f1989 --- /dev/null +++ b/tests/integration/dense-intraday-retention.test.ts @@ -0,0 +1,308 @@ +/** + * v1.10.2 — real-Postgres integration coverage for + * `runDenseIntradayRetention()`. The mocked unit tests could not surface + * the v1.10.0.2 P2002 fold-coexistence bug: the fold mints a canonical + * daily-mean row at the user's local-noon instant, but a row may ALREADY + * sit on that `(userId, type, measuredAt, source, sleepStage)` unique + * index from another path. An externalId-keyed upsert misses the lookup + * and the INSERT then violates the measured_at index with P2002. + * + * This file seeds that exact colliding shape against a live Postgres and + * asserts the reworked fold coexists with the pre-existing daily row, + * stays idempotent, and preserves the dense-tier scope. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; +import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; + +const TEST_USER_ID = "user-dense-retention"; +const TZ = "Europe/Berlin"; +const DAY_KEY = "2026-05-01"; +// The canonical local-noon instant the fold writes for the seeded day. +const CANONICAL = canonicalDailyTimestamp(DAY_KEY, TZ); +const HRV_STATS_ID = "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + await getPrismaClient().user.create({ + data: { + id: TEST_USER_ID, + username: "dense-retention", + email: "dense-retention@example.test", + timezone: TZ, + }, + }); +}); + +describe("runDenseIntradayRetention (real Postgres)", () => { + it("folds out-of-window HRV per-sample rows into one daily MEAN and soft-deletes them", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 40, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T08:00:00.000Z"), + externalId: "hk-hrv-1", + }, + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 60, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T09:00:00.000Z"), + externalId: "hk-hrv-2", + }, + ], + }); + + const summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + + expect(summary.totals.daysConsolidated).toBe(1); + expect(summary.totals.perSampleRowsSoftDeleted).toBe(2); + + const live = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + }); + expect(live).toHaveLength(1); + // MEAN of (40, 60) = 50. + expect(live[0].value).toBeCloseTo(50, 6); + expect(live[0].externalId).toBe(HRV_STATS_ID); + expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); + }); + + it("coexists with a pre-existing daily row on the canonical instant (the P2002 case)", async () => { + const prisma = getPrismaClient(); + // A previously-collapsed daily `stats:` row ALREADY sits on the + // canonical local-noon instant under a DIFFERENT HK stats id than the + // one the fold mints (e.g. a row minted by a sibling consolidation path, + // or a legacy stats id). The scan EXCLUDES it (NOT startsWith 'stats:'), + // so it is never folded into the mean — but pre-v1.10.2 the fold's + // externalId-keyed upsert missed it on lookup and the INSERT then + // collided on (userId, type, measuredAt, source, sleepStage) with P2002. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 999, // stale value to be overwritten by the fold + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: CANONICAL, + externalId: "stats:legacy-hrv-id:2026-05-01", + }, + }); + // Out-of-window per-sample rows that should fold into the canonical row. + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 40, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T06:00:00.000Z"), + externalId: "hk-hrv-1", + }, + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 60, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T18:00:00.000Z"), + externalId: "hk-hrv-2", + }, + ], + }); + + // Must not throw P2002. + const summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + expect(summary.totals.daysConsolidated).toBe(1); + + const live = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + }); + // Exactly one live daily row: the pre-existing one, adopted in place. + expect(live).toHaveLength(1); + expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); + // The fold refreshed the value to the per-sample MEAN (50), overwriting + // the stale 999, and stamped the canonical `stats:` externalId. + expect(live[0].value).toBeCloseTo(50, 6); + expect(live[0].externalId).toBe(HRV_STATS_ID); + + // The two out-of-window samples are tombstoned. + const tombstoned = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: { not: null }, + }, + }); + expect(tombstoned).toHaveLength(2); + }); + + it("does not tombstone a per-sample row that happens to fall on the canonical instant", async () => { + const prisma = getPrismaClient(); + // One per-sample row sits exactly on the canonical local-noon instant; + // it must become the adopted daily row, not get soft-deleted. + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 50, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: CANONICAL, + externalId: "hk-hrv-noon", + }, + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 70, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T06:00:00.000Z"), + externalId: "hk-hrv-morning", + }, + ], + }); + + const summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + expect(summary.totals.daysConsolidated).toBe(1); + + const live = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + }); + expect(live).toHaveLength(1); + expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); + // MEAN of (50, 70) = 60 — the adopted row carries the day's mean. + expect(live[0].value).toBeCloseTo(60, 6); + expect(live[0].externalId).toBe(HRV_STATS_ID); + }); + + it("is idempotent — a second run is a no-op and the value stays correct", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "PULSE", + value: 60, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T06:00:00.000Z"), + externalId: "hk-pulse-1", + }, + { + userId: TEST_USER_ID, + type: "PULSE", + value: 80, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T18:00:00.000Z"), + externalId: "hk-pulse-2", + }, + ], + }); + + const first = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + expect(first.totals.daysConsolidated).toBe(1); + expect(first.totals.perSampleRowsSoftDeleted).toBe(2); + + // Second run must converge to zero work and never collide on the + // canonical row it minted on the first pass. + const second = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + expect(second.totals.daysConsolidated).toBe(0); + expect(second.totals.perSampleRowsSoftDeleted).toBe(0); + + const live = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "PULSE", deletedAt: null }, + }); + expect(live).toHaveLength(1); + // MEAN of (60, 80) = 70. + expect(live[0].value).toBeCloseTo(70, 6); + }); + + it("keeps in-window per-sample rows raw (retention bound)", async () => { + const prisma = getPrismaClient(); + const recent = new Date(Date.now() - 2 * 24 * 60 * 60 * 1000); // 2 days ago + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 45, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: recent, + externalId: "hk-hrv-recent-1", + }, + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 55, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date(recent.getTime() + 60 * 60 * 1000), + externalId: "hk-hrv-recent-2", + }, + ], + }); + + const summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + // Default 14-day window — both rows are inside it. + log: () => {}, + }); + expect(summary.totals.daysConsolidated).toBe(0); + + const live = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + }); + // Both raw samples survive — the intra-day shape is preserved. + expect(live).toHaveLength(2); + }); +});