test(personal-records): detection worker + queue + UI badge coverage

Adds the integration suite for the detection worker against a real
Postgres testcontainer. Covers measurement-driven PRs (steps,
RHR, VO2 max), idempotency on re-run, silent-flag propagation,
and the drift guard that every PR-trackable metric type lands in
the scan loop.

Also patches a pre-existing schema gap: the
(userId, metricType, metricSlot, achievedAt) unique index on
personal_records is NULLS-DISTINCT in Postgres, so the index alone
does not dedup the measurement-driven rows (metricSlot = NULL).
The worker now does an application-level pre-flight findFirst on
the same composite key before each insert, keeping both the
workout path (index covers it) and the measurement path
(application-level guard) idempotent without a Migration.

src/lib/personal-records/__tests__/pr-detection-worker.test.ts

@@ -76,6 +76,7 @@ interface FakeWhere {
   sportType?: string;
   totalDistanceM?: { gte: number };
   durationSec?: { gt: number } | { gte: number };
+  achievedAt?: Date;
 }
 
 function makeFakePrisma(state: {
@@ -122,6 +123,11 @@ function makeFakePrisma(state: {
       row.metricSlot !== where.metricSlot
     )
       return false;
+    if (
+      where.achievedAt !== undefined &&
+      row.achievedAt.getTime() !== where.achievedAt.getTime()
+    )
+      return false;
     return true;
   }
 

src/lib/personal-records/pr-detection-worker.ts

@@ -240,8 +240,23 @@ export async function detectPersonalRecordsForUser(
     const outcome = compareToCurrentBest(best.value, currentPR?.value, direction);
     if (outcome === "no-improvement") continue;
 
-    // ON CONFLICT DO NOTHING via `skipDuplicates`. Re-running the
-    // same job is a guaranteed no-op once the row exists.
+    // Application-level idempotency. The DB unique index covers the
+    // workout case (`metric_slot` non-null) but Postgres' default
+    // NULLS-DISTINCT semantics mean `(userId, metric, NULL, ts)`
+    // doesn't dedup at the index level — a second handler invocation
+    // would otherwise write a second row at the same achievedAt. The
+    // explicit pre-flight findFirst keeps both code paths safe.
+    const existing = await prisma.personalRecord.findFirst({
+      where: {
+        userId,
+        metricType,
+        metricSlot: null,
+        achievedAt: best.measuredAt,
+      },
+      select: { id: true },
+    });
+    if (existing) continue;
+
     const result = await prisma.personalRecord.createMany({
       data: [
         {

tests/integration/pr-detection-end-to-end.test.ts

@@ -0,0 +1,188 @@
+/**
+ * v1.4.25 W16c — PR detection end-to-end.
+ *
+ * The handler runs against a real Postgres testcontainer + real
+ * Measurement / Workout rows. We assert two flows:
+ *
+ *   1. After enough measurements land, the detector writes a
+ *      PersonalRecord row at the all-time best and tags the
+ *      `sourceMeasurementId` to the winning row.
+ *   2. The detector is idempotent — re-running the same handler over
+ *      the same dataset produces no duplicate rows because the
+ *      `(userId, metricType, metricSlot, achievedAt)` unique index
+ *      absorbs the second insert.
+ *
+ * The HTTP-layer enqueue is covered by the per-route unit suites
+ * (`src/app/api/measurements/batch/__tests__/pr-detection-hook.test.ts`
+ * and the workouts equivalent). This file walks the handler instead
+ * of going through pg-boss so the integration suite stays free of
+ * out-of-process job-queue plumbing.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const TEST_USER_ID = "user-pr-detection-e2e";
+
+const { detectPersonalRecordsForUser, PR_DETECTION_WARMUP_THRESHOLD } =
+  await import("@/lib/personal-records/pr-detection-worker");
+const { isPRTrackable } = await import("@/lib/personal-records/pr-direction");
+import type { MeasurementType } from "@/generated/prisma/client";
+const { measurementTypeEnum } = await import("@/lib/validations/measurement");
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+  await getPrismaClient().user.create({
+    data: {
+      id: TEST_USER_ID,
+      username: "pr-detection",
+      email: "pr-detection@example.test",
+    },
+  });
+});
+
+describe("PR detection — measurement-driven (real DB)", () => {
+  it("writes one PersonalRecord row for the all-time best after the warm-up gate", async () => {
+    const prisma = getPrismaClient();
+
+    // Seed enough samples to clear the warm-up gate plus one obvious
+    // all-time best. Days are stepped backwards from today so the
+    // best row's `measuredAt` lands at a deterministic, unique
+    // timestamp.
+    const baseDate = new Date("2026-05-14T00:00:00Z");
+    const samples = [];
+    for (let i = 0; i < PR_DETECTION_WARMUP_THRESHOLD; i++) {
+      samples.push({
+        userId: TEST_USER_ID,
+        type: "ACTIVITY_STEPS" as const,
+        value: 8000 + i * 100,
+        unit: "count",
+        source: "APPLE_HEALTH" as const,
+        measuredAt: new Date(baseDate.getTime() - (i + 1) * 86_400_000),
+        externalId: `seed-${i}`,
+      });
+    }
+    samples.push({
+      userId: TEST_USER_ID,
+      type: "ACTIVITY_STEPS" as const,
+      value: 18234,
+      unit: "count",
+      source: "APPLE_HEALTH" as const,
+      measuredAt: baseDate,
+      externalId: "all-time-best",
+    });
+    await prisma.measurement.createMany({ data: samples });
+
+    const result = await detectPersonalRecordsForUser(TEST_USER_ID);
+    expect(result.inserted).toBeGreaterThanOrEqual(1);
+
+    const stepsPR = await prisma.personalRecord.findFirst({
+      where: {
+        userId: TEST_USER_ID,
+        metricType: "ACTIVITY_STEPS",
+        metricSlot: null,
+      },
+    });
+    expect(stepsPR).not.toBeNull();
+    expect(stepsPR?.value).toBe(18234);
+    expect(stepsPR?.direction).toBe("MAX");
+    expect(stepsPR?.sourceMeasurementId).toBeDefined();
+  });
+
+  it("is idempotent — re-running the handler does not duplicate the row", async () => {
+    const prisma = getPrismaClient();
+
+    const baseDate = new Date("2026-05-14T00:00:00Z");
+    const samples = [];
+    for (let i = 0; i < PR_DETECTION_WARMUP_THRESHOLD; i++) {
+      samples.push({
+        userId: TEST_USER_ID,
+        type: "RESTING_HEART_RATE" as const,
+        value: 65 + i,
+        unit: "bpm",
+        source: "APPLE_HEALTH" as const,
+        measuredAt: new Date(baseDate.getTime() - (i + 1) * 86_400_000),
+        externalId: `rhr-seed-${i}`,
+      });
+    }
+    samples.push({
+      userId: TEST_USER_ID,
+      type: "RESTING_HEART_RATE" as const,
+      value: 47,
+      unit: "bpm",
+      source: "APPLE_HEALTH" as const,
+      measuredAt: baseDate,
+      externalId: "rhr-best",
+    });
+    await prisma.measurement.createMany({ data: samples });
+
+    await detectPersonalRecordsForUser(TEST_USER_ID);
+    const after1 = await prisma.personalRecord.count({
+      where: { userId: TEST_USER_ID, metricType: "RESTING_HEART_RATE" },
+    });
+
+    await detectPersonalRecordsForUser(TEST_USER_ID);
+    const after2 = await prisma.personalRecord.count({
+      where: { userId: TEST_USER_ID, metricType: "RESTING_HEART_RATE" },
+    });
+
+    expect(after1).toBe(1);
+    expect(after2).toBe(1);
+  });
+
+  it("propagates the silent flag without affecting the row written", async () => {
+    const prisma = getPrismaClient();
+    const baseDate = new Date("2026-05-14T00:00:00Z");
+    const samples = [];
+    for (let i = 0; i < PR_DETECTION_WARMUP_THRESHOLD; i++) {
+      samples.push({
+        userId: TEST_USER_ID,
+        type: "VO2_MAX" as const,
+        value: 38 + i * 0.5,
+        unit: "ml/kg/min",
+        source: "APPLE_HEALTH" as const,
+        measuredAt: new Date(baseDate.getTime() - (i + 1) * 86_400_000),
+        externalId: `vo2-${i}`,
+      });
+    }
+    samples.push({
+      userId: TEST_USER_ID,
+      type: "VO2_MAX" as const,
+      value: 52.5,
+      unit: "ml/kg/min",
+      source: "APPLE_HEALTH" as const,
+      measuredAt: baseDate,
+      externalId: "vo2-best",
+    });
+    await prisma.measurement.createMany({ data: samples });
+
+    const result = await detectPersonalRecordsForUser(TEST_USER_ID, {
+      silent: true,
+    });
+    expect(result.silent).toBe(true);
+
+    const pr = await prisma.personalRecord.findFirst({
+      where: { userId: TEST_USER_ID, metricType: "VO2_MAX" },
+    });
+    expect(pr).not.toBeNull();
+    expect(pr?.value).toBeCloseTo(52.5);
+  });
+});
+
+describe("PR detection — drift guard (real DB)", () => {
+  it("scans every PR-trackable metric type when no row exists", async () => {
+    // Drift assertion — `scanned` should match the count of PR-trackable
+    // metrics (and at least every direction enum case must appear in
+    // either the worker or the suppress branch via `getPRDirection`).
+    const result = await detectPersonalRecordsForUser(TEST_USER_ID);
+
+    const trackable = measurementTypeEnum.options.filter((t) =>
+      isPRTrackable(t as MeasurementType),
+    );
+    expect(result.scanned).toBeGreaterThanOrEqual(trackable.length);
+  });
+});

tests/integration/setup.ts

@@ -57,6 +57,7 @@ export async function truncateAllTables(client: PrismaClient): Promise<void> {
     "notification_channels",
     "notification_preferences",
     "passkeys",
+    "personal_records",
     "push_subscriptions",
     "rate_limits",
     "refresh_tokens",