mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 21:10:59 +00:00
merge(WS3PERF): A2-M2 boundary align + Withings/med-intake N+1 fixes
This commit is contained in:
commit
7093e46e41
7 files changed
+455
-110
No files matched your search
@@ -652,4 +652,54 @@ describe("computeSummariesSlice", () => {
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* A2-M2 — read-swap boundary-day consistency. The warm
|
||||
* `computeFromRollups` path windows the regression on UTC-midnight
|
||||
* (`startOfUtcDay(now − N days)` + `composeWindowedRegression`). The
|
||||
* cold-fallback `windowed` query must anchor its REGR_* FILTERs on the
|
||||
* SAME UTC-midnight boundary — `date_trunc('day', NOW() AT TIME ZONE
|
||||
* 'UTC') − INTERVAL 'N days'` — so a warm→cold coverage transition
|
||||
* returns identical boundary-day membership for slope7/30/90 instead of
|
||||
* a cache-dependent answer to the same request.
|
||||
*/
|
||||
describe("A2-M2 regression-window boundary alignment", () => {
|
||||
it("anchors the cold-fallback slope FILTERs on the UTC-midnight day boundary, not the wall-clock instant", async () => {
|
||||
const queries: string[] = [];
|
||||
UNSAFE.mockImplementation((sql: string) => {
|
||||
queries.push(sql);
|
||||
return Promise.resolve([]);
|
||||
});
|
||||
|
||||
// Empty coverage ⇒ cold-fallback path ⇒ the `windowed` query runs.
|
||||
RAW.mockResolvedValueOnce([]);
|
||||
await computeSummariesSlice("user-a2m2");
|
||||
|
||||
const windowedSql = queries.find(
|
||||
(sql) => sql.includes("AS slope7") && sql.includes("REGR_SLOPE"),
|
||||
);
|
||||
expect(windowedSql).toBeDefined();
|
||||
const sql = windowedSql as string;
|
||||
|
||||
// Every regression window (7/30/90 for slope + r²) anchors on the
|
||||
// day-truncated UTC boundary. Six FILTERs total.
|
||||
for (const days of ["7 days", "30 days", "90 days"]) {
|
||||
expect(sql).toContain(
|
||||
`(date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days}') AT TIME ZONE 'UTC'`,
|
||||
);
|
||||
}
|
||||
const anchored = sql.match(
|
||||
/date_trunc\('day', NOW\(\) AT TIME ZONE 'UTC'\) - INTERVAL '(?:7|30|90) days'\) AT TIME ZONE 'UTC'/g,
|
||||
);
|
||||
expect(anchored?.length).toBe(6);
|
||||
|
||||
// The regression windows must NOT fall back to the bare wall-clock
|
||||
// `NOW() - INTERVAL 'N days'` bound the warm path never uses for the
|
||||
// slope. Isolate the slope columns (everything from the first
|
||||
// REGR_SLOPE onward) so the avg7/avg30 windows above — which DO stay
|
||||
// wall-clock by design — don't trip the assertion.
|
||||
const slopeBlock = sql.slice(sql.indexOf("REGR_SLOPE"));
|
||||
expect(slopeBlock).not.toMatch(/WHERE m\."measured_at" >= NOW\(\) -/);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -772,41 +772,54 @@ async function computeFromLiveAggregate(
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
)::double precision AS median,
|
||||
-- A2-M2 — anchor the regression windows on the UTC-midnight day
|
||||
-- boundary so this cold-fallback path returns the SAME boundary-day
|
||||
-- membership the warm computeFromRollups composes via
|
||||
-- startOfUtcDay(now - N days) + composeWindowedRegression. The
|
||||
-- expression date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL
|
||||
-- 'N days' is the SQL twin of startOfUtcDay; the trailing AT TIME
|
||||
-- ZONE 'UTC' re-materialises the naive truncation as a UTC instant
|
||||
-- so the measured_at (timestamptz) comparison lands on the same wall
|
||||
-- the rollup bucketStart sits on. The wall-clock NOW() - INTERVAL
|
||||
-- bound let the boundary day flip in/out of the slope window across
|
||||
-- a warm-to-cold coverage transition (cache-dependent slope on the
|
||||
-- same request); the avg7/30/median windows above stay wall-clock
|
||||
-- because the warm narrow query computes THOSE the same way.
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS slope7,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS r2_7,
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS slope30,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS r2_30,
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS slope90,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS r2_90
|
||||
FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
|
||||
GROUP BY m."type"
|
||||
|
||||
@@ -682,41 +682,54 @@ async function buildFromLiveAggregate(
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '60 days'
|
||||
AND m."measured_at" < NOW() - INTERVAL '30 days'
|
||||
)::double precision AS avg30_last_month,
|
||||
-- A2-M2 — the regression windows anchor on the UTC-midnight day
|
||||
-- boundary, NOT the wall-clock instant, so this cold-fallback path
|
||||
-- returns the SAME boundary-day membership the warm rollup path
|
||||
-- composes from (startOfUtcDay(now - N days) +
|
||||
-- composeWindowedRegression). The expression
|
||||
-- date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL 'N days'
|
||||
-- is the SQL twin of startOfUtcDay; the trailing AT TIME ZONE 'UTC'
|
||||
-- re-materialises the naive truncation as a UTC instant so the
|
||||
-- measured_at (timestamptz) comparison is on the same wall the
|
||||
-- rollup bucketStart sits on. Without this anchor an account that
|
||||
-- crosses a coverage boundary (warm to cold) saw the boundary day
|
||||
-- flip in/out of the slope window, giving a cache-dependent answer
|
||||
-- to the same request.
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS slope7,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS r2_7,
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS slope30,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS r2_30,
|
||||
REGR_SLOPE(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS slope90,
|
||||
REGR_R2(
|
||||
m."value",
|
||||
EXTRACT(EPOCH FROM m."measured_at") / 86400.0
|
||||
) FILTER (
|
||||
WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
|
||||
WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC'
|
||||
)::double precision AS r2_90
|
||||
FROM cm m
|
||||
JOIN window_stats ws ON ws."type" = m."type"
|
||||
|
||||
@@ -177,3 +177,69 @@ describe("dedupeUserIntakeSlots — per-slot error isolation", () => {
|
||||
expect(prisma.medicationIntakeEvent.updateMany).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* v1.20.2 F4 — the per-medication dedup read path is batched. The
|
||||
* pre-v1.20.2 shape issued a `findFirst` (latest takenAt) + a `findMany`
|
||||
* (all rows) PER medication → 2N round-trips. The batched path pulls
|
||||
* every live row for the user's medication set in ONE `findMany`, groups
|
||||
* in memory, and derives the per-med `lastIntakeAt` from the grouped set,
|
||||
* so the read cost is one query regardless of the medication count.
|
||||
*/
|
||||
describe("dedupeUserIntakeSlots — batched read path (F4)", () => {
|
||||
it("reads every medication's rows in a single findMany and never per-med findFirst", async () => {
|
||||
const MED_2 = { ...MED, id: "med-2" };
|
||||
vi.mocked(prisma.medication.findMany).mockResolvedValue([
|
||||
MED,
|
||||
MED_2,
|
||||
] as never);
|
||||
|
||||
// Rows for BOTH medications come back in one ordered result set; the
|
||||
// grouping keys them apart by `medicationId`. med-2 carries a single
|
||||
// row (no duplicate slot) so it is a no-op — proving the grouping
|
||||
// routes each med's slice correctly.
|
||||
const med2Row = {
|
||||
id: "r-m2",
|
||||
medicationId: "med-2",
|
||||
scheduledFor: new Date("2026-06-15T05:00:00.000Z"),
|
||||
takenAt: new Date("2026-06-15T05:02:00.000Z"),
|
||||
skipped: false,
|
||||
syncVersion: 0,
|
||||
createdAt: new Date("2026-06-15T00:00:00Z"),
|
||||
attributionSource: "AUTO",
|
||||
inventoryConsumption: null,
|
||||
};
|
||||
// `inventoryConsumption: null` on every row keeps the loser-refund
|
||||
// path (which itself runs a `findFirst` inside `restoreForIntake`)
|
||||
// out of the picture, so the findFirst assertion isolates the removed
|
||||
// lastIntake probe.
|
||||
const med1Rows = seedRows().map((r) => ({
|
||||
...r,
|
||||
inventoryConsumption: null,
|
||||
}));
|
||||
vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValue([
|
||||
...med1Rows,
|
||||
med2Row,
|
||||
] as never);
|
||||
vi.mocked(prisma.medicationIntakeEvent.update).mockResolvedValue(
|
||||
{} as never,
|
||||
);
|
||||
|
||||
const summary = await dedupeUserIntakeSlots("u1");
|
||||
|
||||
// ONE findMany for the whole medication set; the per-med findFirst is
|
||||
// gone entirely.
|
||||
expect(prisma.medicationIntakeEvent.findMany).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.medicationIntakeEvent.findFirst).not.toHaveBeenCalled();
|
||||
// The single read is keyed by the medication-id set, not per med.
|
||||
const whereArg = vi.mocked(prisma.medicationIntakeEvent.findMany).mock
|
||||
.calls[0][0] as {
|
||||
where: { medicationId: { in: string[] } };
|
||||
};
|
||||
expect(whereArg.where.medicationId.in.sort()).toEqual(["med-1", "med-2"]);
|
||||
|
||||
// med-1's two slots still collapse from its grouped slice; med-2's
|
||||
// lone row is left alone.
|
||||
expect(summary.slotsCollapsed).toBe(2);
|
||||
});
|
||||
});
|
||||
@@ -242,25 +242,24 @@ export async function dedupeUserIntakeSlots(
|
||||
// recompute after the collapse. Deduped so a busy day folds once.
|
||||
const daysToRecompute = new Set<string>();
|
||||
|
||||
for (const med of medications) {
|
||||
if (med.schedules.length === 0) continue;
|
||||
// v1.20.2 F4 — batch the per-medication reads. The pre-v1.20.2 shape
|
||||
// issued a `findFirst` (latest takenAt) + a `findMany` (all live rows)
|
||||
// PER medication → 2N round-trips (20 meds = 40). Pull every live row
|
||||
// for the user's medication set in ONE ordered `findMany`, group in
|
||||
// memory by `medicationId`, and derive the per-med `lastIntakeAt` (max
|
||||
// non-null takenAt) from the same grouped set the `findFirst` would
|
||||
// have scanned. The grouping preserves the `scheduledFor asc` order the
|
||||
// snap + heal passes depend on (a stable Postgres sort keeps the
|
||||
// per-medication slices in the same order a per-med query returned).
|
||||
const medIds = medications
|
||||
.filter((m) => m.schedules.length > 0)
|
||||
.map((m) => m.id);
|
||||
|
||||
// Latest non-tombstoned takenAt — only rolling schedules consult it,
|
||||
// but the snap resolver needs it threaded in.
|
||||
const lastIntake = await prisma.medicationIntakeEvent.findFirst({
|
||||
where: {
|
||||
userId,
|
||||
medicationId: med.id,
|
||||
deletedAt: null,
|
||||
takenAt: { not: null },
|
||||
},
|
||||
orderBy: { takenAt: "desc" },
|
||||
select: { takenAt: true },
|
||||
});
|
||||
const lastIntakeAt = lastIntake?.takenAt ?? null;
|
||||
|
||||
const rows = (await prisma.medicationIntakeEvent.findMany({
|
||||
where: { userId, medicationId: med.id, deletedAt: null },
|
||||
const rowsByMedication = new Map<string, DedupIntakeRow[]>();
|
||||
const lastIntakeAtByMedication = new Map<string, Date | null>();
|
||||
if (medIds.length > 0) {
|
||||
const allRows = (await prisma.medicationIntakeEvent.findMany({
|
||||
where: { userId, medicationId: { in: medIds }, deletedAt: null },
|
||||
select: {
|
||||
id: true,
|
||||
medicationId: true,
|
||||
@@ -276,9 +275,37 @@ export async function dedupeUserIntakeSlots(
|
||||
// v1.16.10 — stamped losers refund their consumption on collapse.
|
||||
inventoryConsumption: true,
|
||||
},
|
||||
orderBy: { scheduledFor: "asc" },
|
||||
// `medicationId` first so the per-med slices are contiguous, then the
|
||||
// `scheduledFor asc` the loop relies on.
|
||||
orderBy: [{ medicationId: "asc" }, { scheduledFor: "asc" }],
|
||||
})) as DedupIntakeRow[];
|
||||
|
||||
for (const row of allRows) {
|
||||
const list = rowsByMedication.get(row.medicationId);
|
||||
if (list) {
|
||||
list.push(row);
|
||||
} else {
|
||||
rowsByMedication.set(row.medicationId, [row]);
|
||||
}
|
||||
// Latest non-tombstoned takenAt per medication — the same value the
|
||||
// per-med `findFirst({ takenAt: { not: null }, orderBy: desc })`
|
||||
// returned. Only rolling schedules consult it, but the snap resolver
|
||||
// needs it threaded in.
|
||||
if (row.takenAt !== null) {
|
||||
const prev = lastIntakeAtByMedication.get(row.medicationId) ?? null;
|
||||
if (prev === null || row.takenAt.getTime() > prev.getTime()) {
|
||||
lastIntakeAtByMedication.set(row.medicationId, row.takenAt);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (const med of medications) {
|
||||
if (med.schedules.length === 0) continue;
|
||||
|
||||
const lastIntakeAt = lastIntakeAtByMedication.get(med.id) ?? null;
|
||||
const rows = rowsByMedication.get(med.id) ?? [];
|
||||
|
||||
if (rows.length < 2) continue;
|
||||
|
||||
// v1.16.9 — heal pass FIRST: converge ad-hoc actioned API/REMINDER
|
||||
|
||||
@@ -7,18 +7,30 @@
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
measurement: {
|
||||
findFirst: vi.fn(),
|
||||
create: vi.fn(),
|
||||
update: vi.fn(),
|
||||
vi.mock("@/lib/db", () => {
|
||||
// v1.20.2 F3 — the write path batches: one `findMany` resolves the
|
||||
// overwrite-vs-insert decision for the whole sync, `createMany` lands
|
||||
// every fresh slot, and changed-value overwrites run inside a
|
||||
// `$transaction([...update])`. The mock `$transaction` awaits the array
|
||||
// of update promises so the per-row `update` mocks still fire and the
|
||||
// call counts assert the batched shape.
|
||||
const measurement = {
|
||||
findMany: vi.fn(),
|
||||
createMany: vi.fn(),
|
||||
update: vi.fn(),
|
||||
};
|
||||
return {
|
||||
prisma: {
|
||||
measurement,
|
||||
withingsConnection: {
|
||||
findUnique: vi.fn(),
|
||||
},
|
||||
$transaction: vi.fn((ops: unknown) =>
|
||||
Array.isArray(ops) ? Promise.all(ops) : Promise.resolve(),
|
||||
),
|
||||
},
|
||||
withingsConnection: {
|
||||
findUnique: vi.fn(),
|
||||
},
|
||||
},
|
||||
}));
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/integrations/status", () => ({
|
||||
isReauthRequired: vi.fn().mockResolvedValue(false),
|
||||
@@ -155,22 +167,35 @@ describe("fetchWithingsActivity", () => {
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* v1.20.2 F3 — helper: pull the flat `data[]` the batched `createMany`
|
||||
* received (the writer always passes a single call with every fresh
|
||||
* slot). Returns `[]` when `createMany` never fired.
|
||||
*/
|
||||
function createManyRows(): Array<Record<string, unknown>> {
|
||||
const calls = vi.mocked(prisma.measurement.createMany).mock.calls;
|
||||
if (calls.length === 0) return [];
|
||||
return (calls[0][0] as { data: Array<Record<string, unknown>> }).data;
|
||||
}
|
||||
|
||||
describe("syncUserActivity — field mapping + idempotency", () => {
|
||||
it("writes one row per (date, metric) the first time a day shows up", async () => {
|
||||
installFetchMock([
|
||||
{ date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
|
||||
]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
// No existing slot rows → every planned row is an insert.
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 3,
|
||||
} as never);
|
||||
|
||||
const imported = await syncUserActivity("user-1");
|
||||
|
||||
expect(imported).toBe(3);
|
||||
expect(prisma.measurement.create).toHaveBeenCalledTimes(3);
|
||||
// v1.20.2 F3 — one batched insert carrying all three metric rows.
|
||||
expect(prisma.measurement.createMany).toHaveBeenCalledTimes(1);
|
||||
|
||||
const types = vi
|
||||
.mocked(prisma.measurement.create)
|
||||
.mock.calls.map((c) => (c[0].data as { type: string }).type);
|
||||
const types = createManyRows().map((r) => r.type as string);
|
||||
expect(types.sort()).toEqual([
|
||||
"ACTIVE_ENERGY_BURNED",
|
||||
"ACTIVITY_STEPS",
|
||||
@@ -178,33 +203,67 @@ describe("syncUserActivity — field mapping + idempotency", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it("updates instead of inserting when the same (date, metric) already exists", async () => {
|
||||
it("updates instead of inserting when the same (date, metric) already exists with a different value", async () => {
|
||||
installFetchMock([
|
||||
{ date: "2026-05-12", steps: 9001, distance: 7000, calories: 450 },
|
||||
]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue({
|
||||
id: "row-existing",
|
||||
} as never);
|
||||
const measuredAt = new Date("2026-05-12T12:00:00.000Z");
|
||||
// Each slot already holds a row with a DIFFERENT value → all three
|
||||
// overwrite (the value moved). Keyed by the same four columns the
|
||||
// batched read matches on.
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([
|
||||
{ id: "row-steps", type: "ACTIVITY_STEPS", measuredAt, value: 1 },
|
||||
{
|
||||
id: "row-distance",
|
||||
type: "WALKING_RUNNING_DISTANCE",
|
||||
measuredAt,
|
||||
value: 1,
|
||||
},
|
||||
{
|
||||
id: "row-calories",
|
||||
type: "ACTIVE_ENERGY_BURNED",
|
||||
measuredAt,
|
||||
value: 1,
|
||||
},
|
||||
] as never);
|
||||
vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
|
||||
|
||||
const imported = await syncUserActivity("user-1");
|
||||
|
||||
expect(imported).toBe(3);
|
||||
expect(prisma.measurement.create).not.toHaveBeenCalled();
|
||||
expect(prisma.measurement.createMany).not.toHaveBeenCalled();
|
||||
// One batched `$transaction` carrying three per-row updates.
|
||||
expect(prisma.$transaction).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.measurement.update).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("skips the write entirely when an existing slot already holds the same value (idempotent re-sync)", async () => {
|
||||
installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
|
||||
const measuredAt = new Date("2026-05-12T12:00:00.000Z");
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([
|
||||
{ id: "row-steps", type: "ACTIVITY_STEPS", measuredAt, value: 8420 },
|
||||
] as never);
|
||||
|
||||
const imported = await syncUserActivity("user-1");
|
||||
|
||||
// The slot is still "touched" (rollup re-fold + imported count), but
|
||||
// no value moved so neither the insert nor the update path runs.
|
||||
expect(imported).toBe(1);
|
||||
expect(prisma.measurement.createMany).not.toHaveBeenCalled();
|
||||
expect(prisma.measurement.update).not.toHaveBeenCalled();
|
||||
expect(prisma.$transaction).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("anchors measuredAt at noon UTC so the instant lands inside the local day for every supported tz", async () => {
|
||||
installFetchMock([{ date: "2026-05-12", steps: 1000 }]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 1,
|
||||
} as never);
|
||||
|
||||
await syncUserActivity("user-1");
|
||||
|
||||
const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
|
||||
data: { measuredAt: Date };
|
||||
};
|
||||
const measuredAt = createArg.data.measuredAt;
|
||||
const measuredAt = createManyRows()[0].measuredAt as Date;
|
||||
expect(measuredAt.toISOString()).toBe("2026-05-12T12:00:00.000Z");
|
||||
|
||||
// Regression: anchoring at noon UTC keeps the row inside the
|
||||
@@ -233,15 +292,15 @@ describe("syncUserActivity — field mapping + idempotency", () => {
|
||||
// `distance` absent — Withings drops the field rather than emitting 0
|
||||
// when there's no GPS. Steps + calories still write.
|
||||
installFetchMock([{ date: "2026-05-12", steps: 8000, calories: 380 }]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 2,
|
||||
} as never);
|
||||
|
||||
const imported = await syncUserActivity("user-1");
|
||||
|
||||
expect(imported).toBe(2);
|
||||
const types = vi
|
||||
.mocked(prisma.measurement.create)
|
||||
.mock.calls.map((c) => (c[0].data as { type: string }).type);
|
||||
const types = createManyRows().map((r) => r.type as string);
|
||||
expect(types).not.toContain("WALKING_RUNNING_DISTANCE");
|
||||
});
|
||||
|
||||
@@ -252,8 +311,10 @@ describe("syncUserActivity — field mapping + idempotency", () => {
|
||||
installFetchMock([
|
||||
{ date: "2026-05-12", steps: 0, distance: 0, calories: 0 },
|
||||
]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 3,
|
||||
} as never);
|
||||
|
||||
const imported = await syncUserActivity("user-1");
|
||||
expect(imported).toBe(3);
|
||||
@@ -261,8 +322,10 @@ describe("syncUserActivity — field mapping + idempotency", () => {
|
||||
|
||||
it("calls recordSyncSuccess after a clean round-trip", async () => {
|
||||
installFetchMock([{ date: "2026-05-12", steps: 1000 }]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 1,
|
||||
} as never);
|
||||
|
||||
await syncUserActivity("user-1");
|
||||
expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
|
||||
@@ -270,18 +333,52 @@ describe("syncUserActivity — field mapping + idempotency", () => {
|
||||
|
||||
it("tags every row with the canonical externalId so future replays dedup", async () => {
|
||||
installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 1,
|
||||
} as never);
|
||||
|
||||
await syncUserActivity("user-1");
|
||||
|
||||
const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
|
||||
data: { externalId: string };
|
||||
};
|
||||
expect(createArg.data.externalId).toBe(
|
||||
expect(createManyRows()[0].externalId).toBe(
|
||||
"withings:activity:user-1:2026-05-12:steps",
|
||||
);
|
||||
});
|
||||
|
||||
it("batches a multi-day backfill into one read and one insert (F3)", async () => {
|
||||
// 5 days × 3 metrics = 15 planned rows. The pre-v1.20.2 shape ran
|
||||
// 15 findFirst + 15 create = 30 round-trips; the batched path is one
|
||||
// findMany + one createMany regardless of the day count.
|
||||
installFetchMock(
|
||||
Array.from({ length: 5 }, (_, i) => ({
|
||||
date: `2026-05-1${i}`,
|
||||
steps: 1000 + i,
|
||||
distance: 2000 + i,
|
||||
calories: 300 + i,
|
||||
})),
|
||||
);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 15,
|
||||
} as never);
|
||||
|
||||
const imported = await syncUserActivity("user-1");
|
||||
|
||||
expect(imported).toBe(15);
|
||||
expect(prisma.measurement.findMany).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.measurement.createMany).toHaveBeenCalledTimes(1);
|
||||
expect(createManyRows()).toHaveLength(15);
|
||||
// The single read is keyed by the full distinct (type, measuredAt)
|
||||
// sets the batch will write — `skipDuplicates` guards the insert.
|
||||
const readArg = vi.mocked(prisma.measurement.findMany).mock.calls[0][0] as {
|
||||
where: { type: { in: unknown[] }; measuredAt: { in: unknown[] } };
|
||||
};
|
||||
expect(readArg.where.type.in).toHaveLength(3);
|
||||
expect(readArg.where.measuredAt.in).toHaveLength(5);
|
||||
const createArg = vi.mocked(prisma.measurement.createMany).mock
|
||||
.calls[0][0] as { skipDuplicates: boolean };
|
||||
expect(createArg.skipDuplicates).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("syncUserActivity — scope-skip guard (v1.4.26)", () => {
|
||||
@@ -300,7 +397,7 @@ describe("syncUserActivity — scope-skip guard (v1.4.26)", () => {
|
||||
|
||||
expect(imported).toBe(0);
|
||||
expect(fetchMock).not.toHaveBeenCalled();
|
||||
expect(prisma.measurement.create).not.toHaveBeenCalled();
|
||||
expect(prisma.measurement.createMany).not.toHaveBeenCalled();
|
||||
expect(recordSyncSuccess).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
@@ -391,8 +488,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => {
|
||||
{ date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
|
||||
{ date: "2026-05-13", steps: 9100, distance: 7100, calories: 430 },
|
||||
]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 0,
|
||||
} as never);
|
||||
|
||||
await syncUserActivity("user-1");
|
||||
|
||||
@@ -418,8 +517,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => {
|
||||
installFetchMock([
|
||||
{ date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 },
|
||||
]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 0,
|
||||
} as never);
|
||||
|
||||
await syncUserActivity("user-1");
|
||||
|
||||
@@ -431,8 +532,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => {
|
||||
// hiccup must never fail the user's sync. The sync's `imported`
|
||||
// return value should still reflect the rows written.
|
||||
installFetchMock([{ date: "2026-05-12", steps: 8420 }]);
|
||||
vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
|
||||
vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
|
||||
vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
|
||||
vi.mocked(prisma.measurement.createMany).mockResolvedValue({
|
||||
count: 0,
|
||||
} as never);
|
||||
vi.mocked(recomputeBucketsForMeasurement).mockRejectedValueOnce(
|
||||
new Error("simulated rollup failure"),
|
||||
);
|
||||
|
||||
@@ -316,6 +316,22 @@ export async function syncUserActivity(
|
||||
// See sync.ts header for the full rationale; the chart's
|
||||
// `source=rollup` fast-path otherwise misses Withings activity days.
|
||||
const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
|
||||
|
||||
// v1.20.2 F3 — flatten (entry × field) into the concrete rows the sync
|
||||
// wants to land, then batch the reads + writes. The pre-v1.20.2 shape
|
||||
// ran a per-(entry, field) `findFirst` + `update`/`create`, i.e.
|
||||
// ~entries × 3 × 2 round-trips (≈ 2,190 on a 365-day backfill). The
|
||||
// overwrite contract is unchanged: an existing WITHINGS row for the
|
||||
// `(type, measuredAt)` slot has its `value` rewritten (the four-column
|
||||
// dedup keys on `(userId, type, measuredAt, source, sleepStage=null)`),
|
||||
// a fresh slot is inserted with the stable `externalId`.
|
||||
interface PlannedRow {
|
||||
type: MeasurementType;
|
||||
measuredAt: Date;
|
||||
value: number;
|
||||
externalId: string;
|
||||
}
|
||||
const planned: PlannedRow[] = [];
|
||||
for (const entry of entries) {
|
||||
if (!entry.date) continue;
|
||||
const measuredAt = activityMeasuredAt(entry.date);
|
||||
@@ -325,42 +341,99 @@ export async function syncUserActivity(
|
||||
// valid (a day of rest still records 0 steps). Only skip
|
||||
// undefined / null / NaN.
|
||||
if (raw == null || !Number.isFinite(raw)) continue;
|
||||
try {
|
||||
const existing = await prisma.measurement.findFirst({
|
||||
where: {
|
||||
userId,
|
||||
type,
|
||||
measuredAt,
|
||||
source: "WITHINGS",
|
||||
sleepStage: null,
|
||||
},
|
||||
select: { id: true },
|
||||
planned.push({
|
||||
type,
|
||||
measuredAt,
|
||||
value: raw,
|
||||
externalId: `withings:activity:${userId}:${entry.date}:${field}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (planned.length > 0) {
|
||||
const slotKey = (type: MeasurementType, measuredAt: Date) =>
|
||||
`${type}|${measuredAt.getTime()}`;
|
||||
|
||||
try {
|
||||
// One read for the whole batch: every existing WITHINGS / non-sleep
|
||||
// row that occupies a slot we intend to write. Keyed by the same
|
||||
// four columns the per-row `findFirst` matched on, so the
|
||||
// overwrite-vs-insert decision is identical — just resolved in
|
||||
// memory instead of N round-trips.
|
||||
const distinctMeasuredAt = [
|
||||
...new Map(planned.map((p) => [p.measuredAt.getTime(), p.measuredAt])),
|
||||
].map(([, d]) => d);
|
||||
const distinctTypes = [...new Set(planned.map((p) => p.type))];
|
||||
const existingRows = await prisma.measurement.findMany({
|
||||
where: {
|
||||
userId,
|
||||
source: "WITHINGS",
|
||||
sleepStage: null,
|
||||
type: { in: distinctTypes },
|
||||
measuredAt: { in: distinctMeasuredAt },
|
||||
},
|
||||
select: { id: true, type: true, measuredAt: true, value: true },
|
||||
});
|
||||
const existingBySlot = new Map<string, { id: string; value: number }>();
|
||||
for (const row of existingRows) {
|
||||
existingBySlot.set(slotKey(row.type, row.measuredAt), {
|
||||
id: row.id,
|
||||
value: row.value,
|
||||
});
|
||||
}
|
||||
|
||||
const toCreate: PlannedRow[] = [];
|
||||
const toUpdate: Array<{ id: string; value: number }> = [];
|
||||
for (const p of planned) {
|
||||
const existing = existingBySlot.get(slotKey(p.type, p.measuredAt));
|
||||
if (existing) {
|
||||
await prisma.measurement.update({
|
||||
where: { id: existing.id },
|
||||
data: { value: raw },
|
||||
});
|
||||
// Only write when the value actually moved — a re-sync of an
|
||||
// unchanged day is then a pure no-op on the write path.
|
||||
if (existing.value !== p.value) {
|
||||
toUpdate.push({ id: existing.id, value: p.value });
|
||||
}
|
||||
} else {
|
||||
await prisma.measurement.create({
|
||||
data: {
|
||||
userId,
|
||||
type,
|
||||
value: raw,
|
||||
unit: getUnitForType(type),
|
||||
measuredAt,
|
||||
source: "WITHINGS",
|
||||
externalId: `withings:activity:${userId}:${entry.date}:${field}`,
|
||||
},
|
||||
});
|
||||
toCreate.push(p);
|
||||
}
|
||||
touched.push({ type, measuredAt });
|
||||
touched.push({ type: p.type, measuredAt: p.measuredAt });
|
||||
imported++;
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(
|
||||
`Failed to upsert activity row (${type}, ${entry.date}): ${err}`,
|
||||
}
|
||||
|
||||
// One insert for every fresh slot. `skipDuplicates` tolerates a
|
||||
// concurrent sibling sync landing the same slot between the read
|
||||
// and this write (the four-column unique catches it).
|
||||
if (toCreate.length > 0) {
|
||||
await prisma.measurement.createMany({
|
||||
data: toCreate.map((p) => ({
|
||||
userId,
|
||||
type: p.type,
|
||||
value: p.value,
|
||||
unit: getUnitForType(p.type),
|
||||
measuredAt: p.measuredAt,
|
||||
source: "WITHINGS" as const,
|
||||
externalId: p.externalId,
|
||||
})),
|
||||
skipDuplicates: true,
|
||||
});
|
||||
}
|
||||
|
||||
// Values differ per row, so the overwrites can't collapse into one
|
||||
// `updateMany`; run them in a single transaction so the batch is one
|
||||
// round-trip rather than one per row.
|
||||
if (toUpdate.length > 0) {
|
||||
await prisma.$transaction(
|
||||
toUpdate.map((u) =>
|
||||
prisma.measurement.update({
|
||||
where: { id: u.id },
|
||||
data: { value: u.value },
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(
|
||||
`Failed to batch-upsert ${planned.length} activity rows for ${userId}: ${err}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user