fix(measurements): rework dense intra-day fold to coexist with the canonical daily row

The dense intra-day retention fold minted its daily-mean stats: row with
an externalId-keyed upsert. On real data a row already occupies the
canonical local-noon instant — a sibling daily stats: row or a per-sample
HRV/PULSE reading that fell on noon — so the upsert lookup missed (the
existing row carries a different externalId) and the INSERT then violated
the (user_id, type, measured_at, source, sleep_stage) unique index with
P2002. The mocked unit tests could not surface it.

Resolve the canonical row by the measured_at composite the collision is on
(sleep_stage IS NULL for these continuous types, matched via the
NULLS NOT DISTINCT index) and adopt it in place — refresh the mean, stamp
the canonical stats: externalId, un-tombstone — or create it when absent.
Exclude the adopted row id from the per-sample soft-delete so a noon-instant
sample becomes the daily row instead of being tombstoned.

Add a testcontainer integration test that seeds the colliding shape (a
pre-existing canonical-instant row plus out-of-window per-sample rows) and
pins no-P2002, coexistence, the noon-sample case, idempotency, and the
retention bound. Update the unit coverage for the adopt-in-place path.
This commit is contained in:
Marc-André Bombeck committed 2026-06-03 07:41:20 +02:00
1 parent e6ade6b03d
commit f049eeb8b1
3 files changed
+434 -36

No files matched your search

@@ -35,14 +35,26 @@ function row(
return { id, type, value, measuredAt: new Date(iso), externalId, unit };
}
function buildPrismaMock(rowsByType: Record<string, unknown[]>) {
const upsert = vi.fn().mockResolvedValue({});
function buildPrismaMock(
rowsByType: Record<string, unknown[]>,
// 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<string, unknown[]>) {
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();
});
});
@@ -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;