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

/**
 * 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 { pulseDayMeanStoredRunsSchema } 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"),
  };
}

// One instance through the cases in order: the readers, the dry run, the
// rebuild after apply (which reads the dry run's record) and the large
// account last, whose rows every earlier count leaves out. The order holds
// under a shuffled run.
describe("the pulse day-mean method, one instance", { shuffle: false }, () => {
  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 = pulseDayMeanStoredRunsSchema.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);
      // The rebuild is on the record beside the dry run, with what it did.
      const { readPulseDayMeanStatus } =
        await import("@/lib/jobs/pulse-day-mean");
      const status = await readPulseDayMeanStatus();
      expect(status.lastRebuild).toMatchObject({ status: "done" });
      expect(status.lastRebuild?.accounts).toBeGreaterThan(0);
      expect(status.lastDryRun?.status).toBe("done");

      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;
    /**
     * The seed's transactions run past Prisma's 5 s default under a loaded
     * host: a month took at most 0.55 s on a quiet machine and 9.9 s at a load
     * average near 95, which expired the transaction. The bound is the
     * fixture's, not the code's under test.
     */
    const SEED_TRANSACTION = { maxWait: 30_000, timeout: 120_000 };

    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.
      //
      // Written a month at a time, each month in its own transaction with the
      // statement timeout lifted for that transaction only and a transaction
      // timeout sized for a loaded host (`SEED_TRANSACTION`): as one statement
      // the million-row insert outran the timeout on CI's shared Postgres, and
      // both timeouts guard the code under test, not this fixture.
      const end = new Date();
      end.setUTCHours(0, 0, 0, 0);
      const start = new Date(end.getTime() - YEARS * 365 * DAY);
      const CHUNK = 30 * DAY;
      for (let from = start.getTime(); from < end.getTime(); from += CHUNK) {
        const chunkFrom = new Date(from);
        const chunkTo = new Date(Math.min(from + CHUNK, end.getTime()));
        await prisma.$transaction(async (tx) => {
          await tx.$executeRaw`SET LOCAL statement_timeout = 0`;
          await tx.$executeRaw`
          INSERT INTO measurements (id, user_id, type, unit, source, value, measured_at, created_at, updated_at)
          SELECT 'big-' || (extract(epoch FROM ts)::bigint / 60)::text, ${BIG}, 'PULSE', 'bpm', 'APPLE_HEALTH',
                 60 + ((extract(epoch FROM ts)::bigint / 60) % 40),
                 ts, now(), now()
          FROM generate_series(
                 ${chunkFrom}::timestamptz,
                 ${chunkTo}::timestamptz - interval '1 minute',
                 interval '1 minute'
               ) AS s(ts)
          WHERE extract(hour FROM ts)::int % 7 <> 3 OR extract(minute FROM ts)::int = 0
        `;
        }, SEED_TRANSACTION);
      }
      await prisma.$transaction(async (tx) => {
        await tx.$executeRaw`SET LOCAL statement_timeout = 0`;
        await tx.$executeRaw`ANALYZE measurements`;
      }, SEED_TRANSACTION);
    }, 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);
  });
});