test(pulse): prove the day-mean method across readers, the dry run's single write and the bounded heap

src/app/api/admin/maintenance/pulse-day-mean/__tests__/route.test.ts

@@ -167,7 +167,10 @@ describe("POST dry_run", () => {
     expect(send).toHaveBeenCalledWith(
       "pulse-day-mean",
       { action: "dry_run" },
-      expect.objectContaining({ retryLimit: 0, singletonKey: "pulse-day-mean" }),
+      expect.objectContaining({
+        retryLimit: 0,
+        singletonKey: "pulse-day-mean",
+      }),
     );
     expect(prisma.appSettings.upsert).toHaveBeenCalledWith(
       expect.objectContaining({

src/lib/jobs/pulse-day-mean.ts

@@ -175,7 +175,10 @@ function windows(
   const start = Math.floor(first.getTime() / DAY_MS) * DAY_MS;
   const end = (Math.floor(last.getTime() / DAY_MS) + 1) * DAY_MS;
   for (let t = start; t < end; t += days * DAY_MS) {
-    out.push({ from: new Date(t), to: new Date(Math.min(end, t + days * DAY_MS)) });
+    out.push({
+      from: new Date(t),
+      to: new Date(Math.min(end, t + days * DAY_MS)),
+    });
   }
   return out;
 }

src/lib/measurements/__tests__/day-mean.test.ts

@@ -175,8 +175,9 @@ describe("pulse day-mean method", () => {
   });
 
   it("weighs hours by min(1, n / 3)", () => {
-    expect([1, 2, 3, 10].map((n) => hourWeight(n, "weighted_hour_mean_v2")))
-      .toEqual([1 / 3, 2 / 3, 1, 1]);
+    expect(
+      [1, 2, 3, 10].map((n) => hourWeight(n, "weighted_hour_mean_v2")),
+    ).toEqual([1 / 3, 2 / 3, 1, 1]);
     expect([1, 2, 3].map((n) => hourWeight(n, "hour_mean_v1"))).toEqual([
       1, 1, 1,
     ]);

src/lib/validations/pulse-day-mean.ts

@@ -14,7 +14,12 @@ export const pulseDayMeanActionSchema = z.strictObject({
 export type PulseDayMeanActionInput = z.infer<typeof pulseDayMeanActionSchema>;
 
 /** Where a day's change between the two methods falls, in bpm. */
-export const PULSE_DAY_MEAN_CHANGE_BUCKETS = ["0-1", "1-2", "2-5", "5+"] as const;
+export const PULSE_DAY_MEAN_CHANGE_BUCKETS = [
+  "0-1",
+  "1-2",
+  "2-5",
+  "5+",
+] as const;
 
 /**
  * What a dry run found. Accounts are numbered in the order they were created

tests/integration/pulse-day-mean-method.test.ts

@@ -0,0 +1,414 @@
+/**
+ * The pulse day-mean method against a real Postgres (#1107 follow-up).
+ *
+ * The worked example: one sparse hour (a single reading at 120 bpm) and 23
+ * dense hours (six readings each at 60 bpm). Every hour weighing one gives
+ * (120 + 23 x 60) / 24 = 62.5; an hour weighing min(1, n / 3) gives
+ * (120 / 3 + 23 x 60) / (23 + 1 / 3) = 60.86.
+ *
+ *   1. Every SQL reader that folds a day from its hours follows the active
+ *      method: the chart's live per-day read, the day view's per-day cells,
+ *      the per-day aggregates, the doctor report's dense buckets, the window
+ *      helpers, and the TypeScript helper beside them.
+ *   2. The dry run writes its result and nothing else: every other table is
+ *      the same, row for row, before and after, the rollup tier included.
+ *   3. The rebuild after apply rewrites the PULSE DAY rollups with the stored
+ *      method.
+ *   4. A large account stays well under the production heap through the dry
+ *      run and the rebuild, because both work a window at a time.
+ */
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import type { Prisma } from "@/generated/prisma/client";
+import { readLocalDailyCells } from "@/lib/day/daily-stats";
+import { loadDenseMeasurementBuckets } from "@/lib/doctor-report/dense-buckets";
+import {
+  computePulseDayMeanDryRun,
+  handlePulseDayMean,
+  type PulseDayMeanPayload,
+} from "@/lib/jobs/pulse-day-mean";
+import { readDayAggregates } from "@/lib/measurements/day-aggregates";
+import { dayValue } from "@/lib/measurements/day-mean";
+import {
+  setActivePulseDayMeanMethod,
+  type PulseDayMeanMethod,
+} from "@/lib/measurements/day-statistic";
+import { readLiveBuckets } from "@/lib/measurements/daily-series-read";
+import { readCanonicalRollupBuckets } from "@/lib/rollups/measurement-read";
+import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups";
+import { invalidateUserTimezone } from "@/lib/tz/resolver";
+import { pulseDayMeanDryRunSchema } from "@/lib/validations/pulse-day-mean";
+import type { Job } from "pg-boss";
+
+const prisma = getPrismaClient();
+const DAY = 86_400_000;
+const HOUR = 3_600_000;
+
+const A = "pulse-method-a";
+const B = "pulse-method-b";
+
+/** Two weeks back, 00:00 UTC: the sparse-hour day; the next day is dense. */
+const now = new Date();
+const D1 = Date.UTC(
+  now.getUTCFullYear(),
+  now.getUTCMonth(),
+  now.getUTCDate() - 14,
+);
+const D2 = D1 + DAY;
+
+const V1 = 62.5;
+const V2 = (120 / 3 + 23 * 60) / (23 + 1 / 3);
+
+function row(
+  userId: string,
+  id: string,
+  value: number,
+  at: number,
+): Prisma.MeasurementCreateManyInput {
+  return {
+    id,
+    userId,
+    type: "PULSE",
+    unit: "bpm",
+    source: "APPLE_HEALTH",
+    value,
+    measuredAt: new Date(at),
+  };
+}
+
+function sparseHourDay(userId: string, day: number) {
+  const rows = [row(userId, `${userId}-${day}-sparse`, 120, day + 30 * 60_000)];
+  for (let h = 1; h < 24; h++) {
+    for (let i = 0; i < 6; i++) {
+      rows.push(
+        row(
+          userId,
+          `${userId}-${day}-${h}-${i}`,
+          60,
+          day + h * HOUR + i * 600_000,
+        ),
+      );
+    }
+  }
+  return rows;
+}
+
+function denseDay(userId: string, day: number) {
+  const rows: Prisma.MeasurementCreateManyInput[] = [];
+  for (let h = 0; h < 24; h++) {
+    for (let i = 0; i < 6; i++) {
+      rows.push(
+        row(
+          userId,
+          `${userId}-${day}-d${h}-${i}`,
+          70,
+          day + h * HOUR + i * 600_000,
+        ),
+      );
+    }
+  }
+  return rows;
+}
+
+const job = (action: PulseDayMeanPayload["action"]) =>
+  [
+    { id: `job-${action}`, data: { action } },
+  ] as unknown as Job<PulseDayMeanPayload>[];
+
+beforeAll(async () => {
+  await truncateAllTables(prisma);
+  setActivePulseDayMeanMethod("hour_mean_v1");
+  await prisma.appSettings.create({ data: { id: "singleton" } });
+  // Created in order: A is account #1, B account #2.
+  await prisma.user.create({
+    data: {
+      id: A,
+      username: A,
+      timezone: "UTC",
+      createdAt: new Date(D1 - 30 * DAY),
+    },
+  });
+  await prisma.user.create({
+    data: {
+      id: B,
+      username: B,
+      timezone: "UTC",
+      createdAt: new Date(D1 - 20 * DAY),
+    },
+  });
+  invalidateUserTimezone(A);
+  invalidateUserTimezone(B);
+  // B: one hour of one reading at 100, one hour of three at 60.
+  // Every hour equal: 80. Weighted: (100 / 3 + 60) / (4 / 3) = 70.
+  await prisma.measurement.createMany({
+    data: [
+      ...sparseHourDay(A, D1),
+      ...denseDay(A, D2),
+      row(B, "b-1", 100, D1 + 8 * HOUR),
+      row(B, "b-2", 60, D1 + 9 * HOUR),
+      row(B, "b-3", 60, D1 + 9 * HOUR + 600_000),
+      row(B, "b-4", 60, D1 + 9 * HOUR + 1_200_000),
+    ],
+  });
+  for (const id of [A, B]) {
+    await recomputeUserRollups(id, {
+      types: ["PULSE"],
+      from: new Date(D1 - 7 * DAY),
+      to: new Date(D2 + 7 * DAY),
+    });
+  }
+}, 120_000);
+
+afterAll(() => setActivePulseDayMeanMethod("hour_mean_v1"));
+
+/** The sparse-hour day as each reader sees it, under `method`. */
+async function readersOfD1(method: PulseDayMeanMethod) {
+  setActivePulseDayMeanMethod(method);
+  const live = await readLiveBuckets({
+    userId: A,
+    type: "PULSE",
+    from: new Date(D1),
+    to: new Date(D2 + DAY - 1),
+    cap: 10,
+    priorityJson: null,
+    grain: "daily",
+    timeZone: "UTC",
+  });
+  const cells = await readLocalDailyCells({
+    userId: A,
+    types: ["PULSE"],
+    from: new Date(D1),
+    to: new Date(D2 + DAY),
+    tz: "UTC",
+    priorityJson: null,
+  });
+  const aggregates = await readDayAggregates({
+    userId: A,
+    type: "PULSE",
+    since: new Date(D1),
+    timeZone: "UTC",
+  });
+  const dense = await loadDenseMeasurementBuckets({
+    userId: A,
+    start: new Date(D1),
+    end: new Date(D1 + DAY - 1),
+    reportTz: "UTC",
+    denseTypes: ["PULSE"],
+  });
+  const rows = await prisma.measurement.findMany({
+    where: { userId: A, measuredAt: { gte: new Date(D1), lt: new Date(D2) } },
+    select: { value: true, measuredAt: true },
+  });
+  const dayKey = new Date(D1).toISOString().slice(0, 10);
+  const agg = aggregates.find((r) => r.day === dayKey)!;
+  return {
+    live: live.find((r) => new Date(r.measuredAt).getTime() === D1)?.value,
+    cells: cells.get("PULSE")?.get(dayKey)?.value,
+    aggregates: agg.dayMean,
+    dense: dense[0].hourWeightedSum! / dense[0].hourWeight!,
+    helper: dayValue("PULSE", rows, "UTC"),
+  };
+}
+
+describe("every reader follows the active method", () => {
+  it("weighs every hour alike under hour_mean_v1", async () => {
+    const seen = await readersOfD1("hour_mean_v1");
+    for (const [reader, value] of Object.entries(seen)) {
+      expect(value, reader).toBeCloseTo(V1, 6);
+    }
+  });
+
+  it("weighs the lone reading's hour a third under weighted_hour_mean_v2", async () => {
+    const seen = await readersOfD1("weighted_hour_mean_v2");
+    for (const [reader, value] of Object.entries(seen)) {
+      expect(value, reader).toBeCloseTo(V2, 6);
+    }
+    expect(Number(V2.toFixed(1))).toBe(60.9);
+    setActivePulseDayMeanMethod("hour_mean_v1");
+  });
+});
+
+/** Every table except the settings row, as one digest per table. */
+async function snapshotTables(): Promise<Record<string, string>> {
+  const tables = await prisma.$queryRaw<Array<{ name: string }>>`
+    SELECT table_name AS name
+    FROM information_schema.tables
+    WHERE table_schema = 'public'
+      AND table_type = 'BASE TABLE'
+      AND table_name NOT IN ('app_settings', '_prisma_migrations')
+    ORDER BY table_name
+  `;
+  const out: Record<string, string> = {};
+  for (const { name } of tables) {
+    if (!/^[a-z0-9_]+$/.test(name)) throw new Error(`table name ${name}`);
+    const [r] = await prisma.$queryRawUnsafe<Array<{ n: number; d: string }>>(
+      `SELECT count(*)::int AS n,
+              coalesce(md5(string_agg(t::text, '|' ORDER BY t::text)), '') AS d
+       FROM "${name}" t`,
+    );
+    out[name] = `${r.n}:${r.d}`;
+  }
+  return out;
+}
+
+/** The settings row without the dry-run column. */
+async function settingsWithoutDryRun() {
+  const [r] = await prisma.$queryRaw<Array<{ j: Record<string, unknown> }>>`
+    SELECT to_jsonb(s) - 'pulse_day_mean_dry_run_json' - 'updated_at' AS j
+    FROM app_settings s WHERE s.id = 'singleton'
+  `;
+  return r.j;
+}
+
+describe("the dry run", () => {
+  it("writes its result and nothing else", async () => {
+    const tablesBefore = await snapshotTables();
+    const settingsBefore = await settingsWithoutDryRun();
+    expect(tablesBefore.measurement_rollups).not.toMatch(/^0:/);
+
+    const outcome = await handlePulseDayMean(job("dry_run"));
+    expect(outcome.ok).toBe(true);
+
+    expect(await snapshotTables()).toEqual(tablesBefore);
+    expect(await settingsWithoutDryRun()).toEqual(settingsBefore);
+
+    const stored = await prisma.appSettings.findUniqueOrThrow({
+      where: { id: "singleton" },
+    });
+    const run = pulseDayMeanDryRunSchema.parse(stored.pulseDayMeanDryRunJson);
+    expect(run.status).toBe("done");
+    expect(run.finishedAt).not.toBeNull();
+    expect(run.result).toEqual({
+      accounts: 2,
+      daysCompared: 3,
+      daysChangedOver1Bpm: 2,
+      maxChangeBpm: 10,
+      distribution: [
+        { bucket: "0-1", days: 1 },
+        { bucket: "1-2", days: 1 },
+        { bucket: "2-5", days: 0 },
+        { bucket: "5+", days: 1 },
+      ],
+      perAccount: [
+        { account: "#1", days: 2, changedOver1Bpm: 1 },
+        { account: "#2", days: 1, changedOver1Bpm: 1 },
+      ],
+    });
+    // No account id leaves the job.
+    expect(JSON.stringify(stored.pulseDayMeanDryRunJson)).not.toContain(A);
+    expect(stored.pulseDayMeanMethod).toBe("hour_mean_v1");
+  });
+});
+
+describe("the rebuild after apply", () => {
+  const dayMean = async (userId: string, day: number) =>
+    (
+      await readCanonicalRollupBuckets({
+        userId,
+        type: "PULSE",
+        granularity: "DAY",
+        from: new Date(day),
+        to: new Date(day),
+        toInclusive: true,
+      })
+    )[0]?.mean;
+
+  it("rewrites the PULSE rollups under the stored method", async () => {
+    expect(await dayMean(A, D1)).toBeCloseTo(V1, 6);
+    expect(await dayMean(B, D1)).toBeCloseTo(80, 6);
+
+    await prisma.appSettings.update({
+      where: { id: "singleton" },
+      data: { pulseDayMeanMethod: "weighted_hour_mean_v2" },
+    });
+    const outcome = await handlePulseDayMean(job("rebuild"));
+    expect(outcome.ok).toBe(true);
+
+    expect(await dayMean(A, D1)).toBeCloseTo(V2, 6);
+    expect(await dayMean(A, D2)).toBeCloseTo(70, 6);
+    expect(await dayMean(B, D1)).toBeCloseTo(70, 6);
+
+    await prisma.appSettings.update({
+      where: { id: "singleton" },
+      data: { pulseDayMeanMethod: "hour_mean_v1" },
+    });
+  });
+});
+
+describe("a large account under the production heap", () => {
+  const BIG = "pulse-method-big";
+  const YEARS = 2;
+  /** The production container's heap ceiling. */
+  const PROD_HEAP_MB = 524;
+
+  beforeAll(async () => {
+    await prisma.user.create({
+      data: { id: BIG, username: BIG, timezone: "UTC" },
+    });
+    // One reading a minute for two years, about a million rows, and every
+    // seventh hour holding a single reading so the two methods disagree.
+    await prisma.$executeRaw`
+      INSERT INTO measurements (id, user_id, type, unit, source, value, measured_at, created_at, updated_at)
+      SELECT 'big-' || g::text, ${BIG}, 'PULSE', 'bpm', 'APPLE_HEALTH',
+             60 + (g % 40),
+             ts, now(), now()
+      FROM generate_series(
+             date_trunc('day', now()) - (${YEARS} * interval '365 days'),
+             date_trunc('day', now()) - interval '1 minute',
+             interval '1 minute'
+           ) WITH ORDINALITY AS s(ts, g)
+      WHERE extract(hour FROM ts)::int % 7 <> 3 OR extract(minute FROM ts)::int = 0
+    `;
+    await prisma.$executeRaw`ANALYZE measurements`;
+  }, 600_000);
+
+  async function peakHeapMb(work: () => Promise<unknown>) {
+    const base = process.memoryUsage().heapUsed;
+    let peak = base;
+    const timer = setInterval(() => {
+      peak = Math.max(peak, process.memoryUsage().heapUsed);
+    }, 25);
+    try {
+      await work();
+    } finally {
+      clearInterval(timer);
+    }
+    peak = Math.max(peak, process.memoryUsage().heapUsed);
+    return {
+      grewMb: (peak - base) / 1_048_576,
+      peakMb: peak / 1_048_576,
+    };
+  }
+
+  it("stays flat through the dry run and the rebuild", async () => {
+    const rows = await prisma.measurement.count({ where: { userId: BIG } });
+    expect(rows).toBeGreaterThan(900_000);
+
+    const dry = await peakHeapMb(() => computePulseDayMeanDryRun());
+    await prisma.appSettings.update({
+      where: { id: "singleton" },
+      data: { pulseDayMeanMethod: "weighted_hour_mean_v2" },
+    });
+    const rebuild = await peakHeapMb(() => handlePulseDayMean(job("rebuild")));
+    process.stderr.write(
+      `[pulse-day-mean heap] rows=${rows} dryRun grew=${dry.grewMb.toFixed(1)} MB peak=${dry.peakMb.toFixed(1)} MB; rebuild grew=${rebuild.grewMb.toFixed(1)} MB peak=${rebuild.peakMb.toFixed(1)} MB\n`,
+    );
+    // A window's aggregate rows, never the readings: a few MB at most. The
+    // bound leaves room for the test runner's own churn and is still a small
+    // slice of the production ceiling.
+    expect(dry.grewMb).toBeLessThan(64);
+    expect(rebuild.grewMb).toBeLessThan(64);
+    expect(dry.grewMb).toBeLessThan(PROD_HEAP_MB / 8);
+
+    const days = await readCanonicalRollupBuckets({
+      userId: BIG,
+      type: "PULSE",
+      granularity: "DAY",
+      from: new Date(Date.now() - 400 * DAY),
+      to: new Date(),
+      toInclusive: true,
+    });
+    expect(days.length).toBeGreaterThan(300);
+  }, 900_000);
+});