feat(pulse): weigh sparse hours in the pulse day mean by the stored method

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

@@ -3,30 +3,70 @@
  * method (#1107 follow-up).
  *
  * GET answers the active method, when it was applied and the last dry run
- * (status and result: per-account labels only, never an id). POST
+ * (status and result: per-account ordinal labels only, never an id). POST
  * `{ action: "dry_run" | "apply" }` answers 202 `{ enqueued }`: a dry run
  * computes both methods for every account and day and writes nothing but its
  * result; `apply` needs a successful dry run from the last seven days (409
  * `maintenance.pulseDayMean.dryRunRequired`), switches the method and
  * rebuilds the pulse rollups. 409 `maintenance.pulseDayMean.busy` while a run
- * is active. Admin only, cookie session only.
+ * is active. Admin only, cookie session only. The work itself runs in the
+ * background worker (`src/lib/jobs/pulse-day-mean.ts`).
  */
 import { NextRequest } from "next/server";
 
 import { apiHandler, requireAdmin } from "@/lib/api-handler";
-import { returnAllZodIssues, safeJson } from "@/lib/api-response";
-import { notImplemented } from "@/lib/http/not-implemented";
+import {
+  apiError,
+  apiSuccess,
+  getClientIp,
+  returnAllZodIssues,
+  safeJson,
+} from "@/lib/api-response";
+import { auditLog } from "@/lib/auth/audit";
+import { prisma } from "@/lib/db";
+import { getGlobalBoss } from "@/lib/jobs/boss-instance";
+import {
+  hasFreshDryRun,
+  isPulseDayMeanRunActive,
+  PULSE_DAY_MEAN_QUEUE,
+  PULSE_DAY_MEAN_SEND_OPTIONS,
+  readPulseDayMeanStatus,
+  writeDryRun,
+  type PulseDayMeanPayload,
+} from "@/lib/jobs/pulse-day-mean";
+import { annotate } from "@/lib/logging/context";
+import {
+  setActivePulseDayMeanMethod,
+  type PulseDayMeanMethod,
+} from "@/lib/measurements/day-statistic";
 import { pulseDayMeanActionSchema } from "@/lib/validations/pulse-day-mean";
 
 export const dynamic = "force-dynamic";
 
+/** The method the apply step switches to. */
+const APPLIED_METHOD: PulseDayMeanMethod = "weighted_hour_mean_v2";
+
+function busy() {
+  return apiError("A pulse day-mean run is already queued or running", 409, {
+    errorCode: "maintenance.pulseDayMean.busy",
+  });
+}
+
 export const GET = apiHandler(async () => {
   await requireAdmin();
-  return notImplemented("admin.pulse_day_mean.get");
+  const status = await readPulseDayMeanStatus();
+  annotate({
+    action: { name: "admin.pulse_day_mean.status" },
+    meta: {
+      method: status.method,
+      dry_run_status: status.lastDryRun?.status ?? null,
+    },
+  });
+  return apiSuccess(status);
 });
 
 export const POST = apiHandler(async (request: NextRequest) => {
-  await requireAdmin();
+  const { user } = await requireAdmin();
 
   const { data: body, error: jsonError } = await safeJson(request, {
     maxBytes: 1024,
@@ -34,6 +74,80 @@ export const POST = apiHandler(async (request: NextRequest) => {
   if (jsonError) return jsonError;
   const parsed = pulseDayMeanActionSchema.safeParse(body);
   if (!parsed.success) return returnAllZodIssues(parsed.error, 422);
+  const action = parsed.data.action;
+  annotate({
+    action: { name: "admin.pulse_day_mean.requested" },
+    meta: { pulse_day_mean_action: action },
+  });
 
-  return notImplemented("admin.pulse_day_mean.post");
+  const boss = getGlobalBoss();
+  if (!boss) return apiError("Background worker is not available", 503);
+  if (await isPulseDayMeanRunActive()) return busy();
+
+  if (action === "dry_run") {
+    const previous = (await readPulseDayMeanStatus()).lastDryRun;
+    await writeDryRun({
+      status: "queued",
+      startedAt: new Date().toISOString(),
+      finishedAt: null,
+      result: null,
+    });
+    const payload: PulseDayMeanPayload = { action: "dry_run" };
+    const jobId = await boss.send(
+      PULSE_DAY_MEAN_QUEUE,
+      payload,
+      PULSE_DAY_MEAN_SEND_OPTIONS,
+    );
+    if (jobId === null) {
+      // Another run took the singleton between the check and the send; the
+      // last dry run stays what it was.
+      if (previous) await writeDryRun(previous);
+      return busy();
+    }
+    annotate({ meta: { pulse_day_mean_enqueued: true } });
+    return apiSuccess({ enqueued: true }, 202);
+  }
+
+  if (!hasFreshDryRun(await readPulseDayMeanStatus())) {
+    return apiError(
+      "Run a dry run first; apply needs one finished in the last seven days",
+      409,
+      { errorCode: "maintenance.pulseDayMean.dryRunRequired" },
+    );
+  }
+
+  // The method first, then the rebuild: the rollups read the stored method
+  // inside their own statements, so a rebuild that starts before the switch
+  // would write the old statistic again.
+  const appliedAt = new Date();
+  await prisma.appSettings.upsert({
+    where: { id: "singleton" },
+    create: {
+      id: "singleton",
+      pulseDayMeanMethod: APPLIED_METHOD,
+      pulseDayMeanAppliedAt: appliedAt,
+    },
+    update: {
+      pulseDayMeanMethod: APPLIED_METHOD,
+      pulseDayMeanAppliedAt: appliedAt,
+    },
+  });
+  setActivePulseDayMeanMethod(APPLIED_METHOD);
+  const payload: PulseDayMeanPayload = { action: "rebuild" };
+  const jobId = await boss.send(
+    PULSE_DAY_MEAN_QUEUE,
+    payload,
+    PULSE_DAY_MEAN_SEND_OPTIONS,
+  );
+  const enqueued = jobId !== null;
+
+  await auditLog("admin.pulseDayMean.apply", {
+    userId: user.id,
+    ipAddress: getClientIp(request),
+    details: { method: APPLIED_METHOD, rebuildEnqueued: enqueued },
+  });
+  annotate({
+    meta: { pulse_day_mean_enqueued: enqueued, method: APPLIED_METHOD },
+  });
+  return apiSuccess({ enqueued }, 202);
 });

src/app/api/measurements/series/route.ts

@@ -35,7 +35,9 @@ import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic";
 import {
   dayWeightedRows,
+  hourWeightSql,
   readingsMean,
+  weightedHourMeanSql,
   windowMeanSql,
   zoneDayFrame,
 } from "@/lib/measurements/day-mean";
@@ -485,8 +487,8 @@ export const GET = apiHandler(async (request: NextRequest) => {
     const type = KIND_TO_TYPE[kind];
     const seriesRows = seriesRowsFrom(priorityJson, type, days);
     // A day of a type with an activity-dependent sampling rate (pulse) is the
-    // mean of its local hours' means, each hour once, so a workout hour with
-    // hundreds of times the readings does not outweigh the rest of the day (see
+    // weighted mean of its local hours' means, so a workout hour with hundreds
+    // of times the readings does not outweigh the rest of the day (see
     // `day-statistic.ts`). The hour grain is already one mean per hour.
     const hourWeighted = grain === "day" && usesHourlyMeanDay(type);
     const bucketRows = await prisma.$queryRawUnsafe<
@@ -515,6 +517,7 @@ export const GET = apiHandler(async (request: NextRequest) => {
         SELECT
           date_trunc($2, local_hour)                              AS local_day,
           AVG("value")                                            AS hour_mean,
+          COUNT(*)                                                AS n,
           MIN(COALESCE("value_min", "value"))                     AS min_value,
           MAX(COALESCE("value_max", "value"))                     AS max_value
         FROM localized
@@ -522,7 +525,7 @@ export const GET = apiHandler(async (request: NextRequest) => {
       )
       SELECT
         local_day AT TIME ZONE $3                                AS bucket_start,
-        AVG(hour_mean)::double precision                         AS mean,
+        ${weightedHourMeanSql("hour_mean", hourWeightSql("n"))}::double precision AS mean,
         MIN(min_value)::double precision                         AS min_value,
         MAX(max_value)::double precision                         AS max_value
       FROM hourly