diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 4cc7415fa..55120800a 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -35753,7 +35753,8 @@ components: labs.update.sourceReferenceRangeInvalid; life_event — life_event.invalid, life_event.not_found; managed_profile — managed_profile.guardian.duplicate, managed_profile.guardian.managed_invitee, managed_profile.guardian.required, managed_profile.handover.no_pending, - managed_profile.handover.unknown_guardian, managed_profile.not_found; measurement — + managed_profile.handover.unknown_guardian, managed_profile.not_found; maintenance — + maintenance.pulseDayMean.busy, maintenance.pulseDayMean.dryRunRequired; measurement — measurement.batch.invalid, measurement.batch.source_not_permitted, measurement.batch.too_large, measurement.bulk-delete.invalid, measurement.create.invalid, measurement.create.source_not_permitted, measurement.delete.too_large, measurement.duplicate_timestamp, measurement.restore.invalid, diff --git a/messages/de.json b/messages/de.json index 9fb9a2e08..632dfa127 100644 --- a/messages/de.json +++ b/messages/de.json @@ -8126,6 +8126,48 @@ "started": "Pflege gestartet.", "alreadyQueued": "Ein Pflegelauf ist bereits eingeplant oder läuft.", "startFailed": "Die Pflege konnte nicht gestartet werden." + }, + "pulseDayMean": { + "title": "Puls-Tagesmittel", + "description": "Wie Stunden mit nur wenigen Messwerten in den Tagespuls eingehen.", + "body": "Bisher zählt jede Stunde mit einem Messwert gleich viel, sodass ein einzelner Ausreißer einen ganzen Tag verschieben kann. Die gewichtete Methode lässt eine Stunde ab drei Messwerten voll zählen und einen einzelnen Messwert zu einem Drittel. Starte zuerst den Probelauf, um zu sehen, wie viele Tage sich ändern würden. Beim Anwenden werden die gespeicherten Pulstage im Hintergrund neu berechnet.", + "method": { + "hourly": "Jede Stunde gleich", + "weighted": "Dünne Stunden gewichtet" + }, + "loading": "Puls-Tagesmittel wird geprüft…", + "loadError": "Der Stand des Puls-Tagesmittels konnte nicht gelesen werden.", + "state": { + "idle": "Noch kein Probelauf.", + "queued": "Der Probelauf wartet auf den Hintergrunddienst.", + "running": "Der Probelauf läuft.", + "done": "Probelauf am {at} abgeschlossen.", + "failed": "Der Probelauf wurde abgebrochen. Das App-Protokoll nennt die Ursache; starte ihn erneut." + }, + "result": { + "days": "Verglichene Tage", + "changed": "Tage mit mehr als 1 bpm Änderung", + "max": "Größte Änderung", + "maxValue": "{value} bpm", + "bucket": "{range} bpm", + "account": "Konto {account}", + "accountDays": "{changed} von {days} Tagen ändern sich" + }, + "appliedAt": "Gewichtete Methode angewendet am {at}.", + "dryRun": "Probelauf", + "apply": "Anwenden", + "confirmTitle": "Gewichtetes Puls-Tagesmittel anwenden?", + "confirmBody": "Danach werden die Pulstage aller Konten mit gewichteten dünnen Stunden berechnet. Die gespeicherten Pulstage werden im Hintergrund neu berechnet, was auf einer großen Instanz eine Weile dauern kann.", + "confirm": "Anwenden", + "cancel": "Abbrechen", + "dryRunStarted": "Probelauf gestartet.", + "applied": "Gewichtete Methode angewendet. Die Pulstage werden neu berechnet.", + "appliedNoRebuild": "Gewichtete Methode angewendet. Ein anderer Lauf arbeitet noch; wende sie erneut an, sobald er fertig ist, damit die Pulstage neu berechnet werden.", + "error": { + "busy": "Ein Probelauf oder eine Neuberechnung läuft noch. Versuche es erneut, sobald er fertig ist.", + "dryRunRequired": "Starte zuerst einen Probelauf. Zum Anwenden braucht es einen, der in den letzten sieben Tagen abgeschlossen wurde.", + "generic": "Die Anfrage ist nicht durchgegangen." + } } }, "carrier": "Netzbetreiber", diff --git a/messages/en.json b/messages/en.json index dfaf30239..fecce85f8 100644 --- a/messages/en.json +++ b/messages/en.json @@ -8126,6 +8126,48 @@ "started": "Maintenance started.", "alreadyQueued": "A maintenance run is already queued or running.", "startFailed": "Could not start the maintenance." + }, + "pulseDayMean": { + "title": "Pulse day mean", + "description": "How hours with only a few readings count toward a day's pulse.", + "body": "Up to now every hour with a reading counts the same, so a single stray reading can move a whole day. The weighted method lets an hour count fully from three readings on, and a lone reading a third. Run the dry run first to see how many days would change. Applying recalculates the stored pulse days in the background.", + "method": { + "hourly": "Every hour equal", + "weighted": "Sparse hours weighted" + }, + "loading": "Checking the pulse day mean…", + "loadError": "Could not read the pulse day-mean status.", + "state": { + "idle": "No dry run yet.", + "queued": "The dry run is waiting for the background worker.", + "running": "The dry run is in progress.", + "done": "Dry run finished on {at}.", + "failed": "The dry run stopped. The app log names the cause; start it again." + }, + "result": { + "days": "Days compared", + "changed": "Days moving more than 1 bpm", + "max": "Largest change", + "maxValue": "{value} bpm", + "bucket": "{range} bpm", + "account": "Account {account}", + "accountDays": "{changed} of {days} days move" + }, + "appliedAt": "Weighted method applied on {at}.", + "dryRun": "Dry run", + "apply": "Apply", + "confirmTitle": "Apply the weighted pulse day mean?", + "confirmBody": "Every account's pulse days are then computed with sparse hours weighted. The stored pulse days are recalculated in the background, which can take a while on a large instance.", + "confirm": "Apply", + "cancel": "Cancel", + "dryRunStarted": "Dry run started.", + "applied": "Weighted method applied. The pulse days are being recalculated.", + "appliedNoRebuild": "Weighted method applied. Another run is still working, so apply again once it has finished to recalculate the pulse days.", + "error": { + "busy": "A dry run or recalculation is still running. Try again once it has finished.", + "dryRunRequired": "Run a dry run first. Applying needs one that finished in the last seven days.", + "generic": "The request did not go through." + } } }, "carrier": "Carrier", diff --git a/messages/es.json b/messages/es.json index 33e992c02..dd77b93f3 100644 --- a/messages/es.json +++ b/messages/es.json @@ -8126,6 +8126,48 @@ "started": "Mantenimiento iniciado.", "alreadyQueued": "Ya hay un mantenimiento en cola o en marcha.", "startFailed": "No se pudo iniciar el mantenimiento." + }, + "pulseDayMean": { + "title": "Media diaria del pulso", + "description": "Cómo cuentan las horas con pocas lecturas en el pulso de un día.", + "body": "Hasta ahora, cada hora con una lectura cuenta lo mismo, así que una sola lectura suelta puede mover un día entero. El método ponderado hace que una hora cuente por completo a partir de tres lecturas, y una lectura aislada un tercio. Ejecuta primero la prueba para ver cuántos días cambiarían. Al aplicarlo, los días de pulso guardados se recalculan en segundo plano.", + "method": { + "hourly": "Todas las horas igual", + "weighted": "Horas escasas ponderadas" + }, + "loading": "Comprobando la media diaria del pulso…", + "loadError": "No se pudo leer el estado de la media diaria del pulso.", + "state": { + "idle": "Aún no hay ninguna prueba.", + "queued": "La prueba está esperando al proceso en segundo plano.", + "running": "La prueba está en curso.", + "done": "Prueba terminada el {at}.", + "failed": "La prueba se detuvo. El registro de la app indica la causa; vuelve a iniciarla." + }, + "result": { + "days": "Días comparados", + "changed": "Días que cambian más de 1 bpm", + "max": "Mayor cambio", + "maxValue": "{value} bpm", + "bucket": "{range} bpm", + "account": "Cuenta {account}", + "accountDays": "Cambian {changed} de {days} días" + }, + "appliedAt": "Método ponderado aplicado el {at}.", + "dryRun": "Prueba", + "apply": "Aplicar", + "confirmTitle": "¿Aplicar la media diaria del pulso ponderada?", + "confirmBody": "A partir de entonces, los días de pulso de todas las cuentas se calculan con las horas escasas ponderadas. Los días de pulso guardados se recalculan en segundo plano, lo que puede tardar en una instancia grande.", + "confirm": "Aplicar", + "cancel": "Cancelar", + "dryRunStarted": "Prueba iniciada.", + "applied": "Método ponderado aplicado. Los días de pulso se están recalculando.", + "appliedNoRebuild": "Método ponderado aplicado. Otro proceso sigue en marcha; vuelve a aplicarlo cuando termine para recalcular los días de pulso.", + "error": { + "busy": "Todavía hay una prueba o un recálculo en marcha. Inténtalo de nuevo cuando termine.", + "dryRunRequired": "Ejecuta primero una prueba. Para aplicar hace falta una terminada en los últimos siete días.", + "generic": "La solicitud no se completó." + } } }, "carrier": "Operador", diff --git a/messages/fr.json b/messages/fr.json index 66e399825..08e203bae 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -8126,6 +8126,48 @@ "started": "Maintenance lancée.", "alreadyQueued": "Une maintenance est déjà en file d'attente ou en cours.", "startFailed": "Impossible de lancer la maintenance." + }, + "pulseDayMean": { + "title": "Moyenne journalière du pouls", + "description": "Comment les heures avec peu de mesures comptent dans le pouls d'une journée.", + "body": "Jusqu'ici, chaque heure avec une mesure compte autant, si bien qu'une seule mesure isolée peut déplacer toute une journée. La méthode pondérée fait compter une heure pleinement à partir de trois mesures, et une mesure isolée pour un tiers. Lancez d'abord la simulation pour voir combien de jours changeraient. L'application recalcule en arrière-plan les journées de pouls enregistrées.", + "method": { + "hourly": "Toutes les heures à égalité", + "weighted": "Heures clairsemées pondérées" + }, + "loading": "Vérification de la moyenne journalière du pouls…", + "loadError": "Impossible de lire l'état de la moyenne journalière du pouls.", + "state": { + "idle": "Aucune simulation pour l'instant.", + "queued": "La simulation attend le processus d'arrière-plan.", + "running": "La simulation est en cours.", + "done": "Simulation terminée le {at}.", + "failed": "La simulation s'est arrêtée. Le journal de l'application en indique la cause ; relancez-la." + }, + "result": { + "days": "Jours comparés", + "changed": "Jours qui bougent de plus de 1 bpm", + "max": "Plus grand écart", + "maxValue": "{value} bpm", + "bucket": "{range} bpm", + "account": "Compte {account}", + "accountDays": "{changed} jours sur {days} changent" + }, + "appliedAt": "Méthode pondérée appliquée le {at}.", + "dryRun": "Simulation", + "apply": "Appliquer", + "confirmTitle": "Appliquer la moyenne journalière pondérée du pouls ?", + "confirmBody": "Les journées de pouls de tous les comptes sont alors calculées avec les heures clairsemées pondérées. Les journées enregistrées sont recalculées en arrière-plan, ce qui peut prendre un moment sur une grande instance.", + "confirm": "Appliquer", + "cancel": "Annuler", + "dryRunStarted": "Simulation lancée.", + "applied": "Méthode pondérée appliquée. Les journées de pouls sont en cours de recalcul.", + "appliedNoRebuild": "Méthode pondérée appliquée. Une autre exécution est encore en cours ; appliquez de nouveau une fois terminée pour recalculer les journées de pouls.", + "error": { + "busy": "Une simulation ou un recalcul est encore en cours. Réessayez une fois terminé.", + "dryRunRequired": "Lancez d'abord une simulation. L'application en exige une terminée au cours des sept derniers jours.", + "generic": "La requête n'a pas abouti." + } } }, "carrier": "Opérateur", diff --git a/messages/it.json b/messages/it.json index 2e29bb05c..2068119a0 100644 --- a/messages/it.json +++ b/messages/it.json @@ -8126,6 +8126,48 @@ "started": "Manutenzione avviata.", "alreadyQueued": "Una manutenzione è già in coda o in corso.", "startFailed": "Impossibile avviare la manutenzione." + }, + "pulseDayMean": { + "title": "Media giornaliera del polso", + "description": "Come contano le ore con poche misurazioni nel polso di una giornata.", + "body": "Finora ogni ora con una misurazione conta allo stesso modo, quindi una sola misurazione isolata può spostare un'intera giornata. Il metodo ponderato fa contare un'ora per intero da tre misurazioni in su, e una misurazione isolata per un terzo. Avvia prima la prova per vedere quanti giorni cambierebbero. Applicandolo, i giorni di polso salvati vengono ricalcolati in background.", + "method": { + "hourly": "Tutte le ore uguali", + "weighted": "Ore rade ponderate" + }, + "loading": "Controllo della media giornaliera del polso…", + "loadError": "Impossibile leggere lo stato della media giornaliera del polso.", + "state": { + "idle": "Ancora nessuna prova.", + "queued": "La prova è in attesa del processo in background.", + "running": "La prova è in corso.", + "done": "Prova terminata il {at}.", + "failed": "La prova si è interrotta. Il registro dell'app ne indica la causa; avviala di nuovo." + }, + "result": { + "days": "Giorni confrontati", + "changed": "Giorni che cambiano di oltre 1 bpm", + "max": "Variazione maggiore", + "maxValue": "{value} bpm", + "bucket": "{range} bpm", + "account": "Account {account}", + "accountDays": "Cambiano {changed} giorni su {days}" + }, + "appliedAt": "Metodo ponderato applicato il {at}.", + "dryRun": "Prova", + "apply": "Applica", + "confirmTitle": "Applicare la media giornaliera del polso ponderata?", + "confirmBody": "Da quel momento i giorni di polso di tutti gli account vengono calcolati con le ore rade ponderate. I giorni salvati vengono ricalcolati in background, il che può richiedere un po' su un'istanza grande.", + "confirm": "Applica", + "cancel": "Annulla", + "dryRunStarted": "Prova avviata.", + "applied": "Metodo ponderato applicato. I giorni di polso vengono ricalcolati.", + "appliedNoRebuild": "Metodo ponderato applicato. Un'altra esecuzione è ancora in corso; applica di nuovo quando è terminata per ricalcolare i giorni di polso.", + "error": { + "busy": "Una prova o un ricalcolo è ancora in corso. Riprova quando è terminato.", + "dryRunRequired": "Avvia prima una prova. Per applicare ne serve una terminata negli ultimi sette giorni.", + "generic": "La richiesta non è andata a buon fine." + } } }, "carrier": "Operatore", diff --git a/messages/ko.json b/messages/ko.json index 23d6ec87a..288cdd975 100644 --- a/messages/ko.json +++ b/messages/ko.json @@ -8101,6 +8101,48 @@ "started": "유지 관리를 시작했습니다.", "alreadyQueued": "유지 관리가 이미 대기 중이거나 실행 중입니다.", "startFailed": "유지 관리를 시작할 수 없습니다." + }, + "pulseDayMean": { + "title": "하루 평균 맥박", + "description": "측정값이 적은 시간이 하루 맥박에 반영되는 방식입니다.", + "body": "지금까지는 측정값이 있는 모든 시간이 똑같이 반영되어, 동떨어진 측정값 하나가 하루 전체를 움직일 수 있습니다. 가중 방식에서는 측정값이 세 개 이상인 시간이 온전히 반영되고, 측정값이 하나뿐인 시간은 3분의 1만 반영됩니다. 먼저 시험 실행으로 바뀌는 날이 얼마나 되는지 확인하세요. 적용하면 저장된 맥박 일별 값이 백그라운드에서 다시 계산됩니다.", + "method": { + "hourly": "모든 시간 동일", + "weighted": "드문 시간 가중" + }, + "loading": "하루 평균 맥박을 확인하는 중…", + "loadError": "하루 평균 맥박 상태를 읽을 수 없습니다.", + "state": { + "idle": "아직 시험 실행이 없습니다.", + "queued": "시험 실행이 백그라운드 작업자를 기다리는 중입니다.", + "running": "시험 실행이 진행 중입니다.", + "done": "{at}에 시험 실행이 끝났습니다.", + "failed": "시험 실행이 중단되었습니다. 앱 로그에 원인이 있습니다. 다시 시작하세요." + }, + "result": { + "days": "비교한 날", + "changed": "1 bpm 넘게 바뀌는 날", + "max": "가장 큰 변화", + "maxValue": "{value} bpm", + "bucket": "{range} bpm", + "account": "계정 {account}", + "accountDays": "{days}일 중 {changed}일 변경" + }, + "appliedAt": "{at}에 가중 방식이 적용되었습니다.", + "dryRun": "시험 실행", + "apply": "적용", + "confirmTitle": "가중 하루 평균 맥박을 적용할까요?", + "confirmBody": "이후 모든 계정의 맥박 일별 값이 드문 시간을 가중하여 계산됩니다. 저장된 맥박 일별 값은 백그라운드에서 다시 계산되며, 큰 인스턴스에서는 시간이 걸릴 수 있습니다.", + "confirm": "적용", + "cancel": "취소", + "dryRunStarted": "시험 실행을 시작했습니다.", + "applied": "가중 방식이 적용되었습니다. 맥박 일별 값을 다시 계산하는 중입니다.", + "appliedNoRebuild": "가중 방식이 적용되었습니다. 다른 실행이 아직 진행 중이니, 끝난 뒤 다시 적용하여 맥박 일별 값을 다시 계산하세요.", + "error": { + "busy": "시험 실행이나 재계산이 아직 진행 중입니다. 끝난 뒤 다시 시도하세요.", + "dryRunRequired": "먼저 시험 실행을 하세요. 적용하려면 최근 7일 안에 끝난 시험 실행이 필요합니다.", + "generic": "요청이 처리되지 않았습니다." + } } }, "carrier": "통신사", diff --git a/messages/pl.json b/messages/pl.json index 2f1669774..d738d3cb8 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -8126,6 +8126,48 @@ "started": "Konserwacja uruchomiona.", "alreadyQueued": "Konserwacja jest już w kolejce lub trwa.", "startFailed": "Nie udało się uruchomić konserwacji." + }, + "pulseDayMean": { + "title": "Dzienna średnia tętna", + "description": "Jak godziny z niewieloma pomiarami liczą się do tętna danego dnia.", + "body": "Dotąd każda godzina z pomiarem liczy się tak samo, więc jeden przypadkowy pomiar może przesunąć cały dzień. Metoda ważona sprawia, że godzina liczy się w pełni od trzech pomiarów, a pojedynczy pomiar w jednej trzeciej. Najpierw uruchom próbę, aby zobaczyć, ile dni by się zmieniło. Zastosowanie przelicza zapisane dni tętna w tle.", + "method": { + "hourly": "Każda godzina tak samo", + "weighted": "Rzadkie godziny ważone" + }, + "loading": "Sprawdzanie dziennej średniej tętna…", + "loadError": "Nie udało się odczytać stanu dziennej średniej tętna.", + "state": { + "idle": "Jeszcze nie było próby.", + "queued": "Próba czeka na proces w tle.", + "running": "Próba trwa.", + "done": "Próba zakończona {at}.", + "failed": "Próba została przerwana. Dziennik aplikacji podaje przyczynę; uruchom ją ponownie." + }, + "result": { + "days": "Porównane dni", + "changed": "Dni ze zmianą powyżej 1 bpm", + "max": "Największa zmiana", + "maxValue": "{value} bpm", + "bucket": "{range} bpm", + "account": "Konto {account}", + "accountDays": "Zmienia się {changed} z {days} dni" + }, + "appliedAt": "Metoda ważona zastosowana {at}.", + "dryRun": "Próba", + "apply": "Zastosuj", + "confirmTitle": "Zastosować ważoną dzienną średnią tętna?", + "confirmBody": "Od tej chwili dni tętna wszystkich kont są liczone z ważonymi rzadkimi godzinami. Zapisane dni tętna są przeliczane w tle, co na dużej instancji może chwilę potrwać.", + "confirm": "Zastosuj", + "cancel": "Anuluj", + "dryRunStarted": "Próba uruchomiona.", + "applied": "Metoda ważona zastosowana. Dni tętna są przeliczane.", + "appliedNoRebuild": "Metoda ważona zastosowana. Inny przebieg wciąż trwa; zastosuj ponownie, gdy się zakończy, aby przeliczyć dni tętna.", + "error": { + "busy": "Próba lub przeliczanie wciąż trwa. Spróbuj ponownie, gdy się zakończy.", + "dryRunRequired": "Najpierw uruchom próbę. Do zastosowania potrzebna jest próba zakończona w ciągu ostatnich siedmiu dni.", + "generic": "Żądanie nie zostało zrealizowane." + } } }, "carrier": "Operator", diff --git a/prisma/schema.prisma b/prisma/schema.prisma index b58af23c3..eb6a974d2 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -5343,11 +5343,14 @@ model AppSettings { // every hour alike; `weighted_hour_mean_v2` weighs an hour by // `min(1, n / 3)`. Switched only by the admin apply step after a dry run; // the dry-run result is the only thing a dry run writes. - /// @pending-consumer(#v143-B8) Active pulse day-mean method (CHECK in migration 0407). + /// Active pulse day-mean method (CHECK in migration 0407). Readers: every + /// process's mirror (`pulse-day-mean-method.ts`) and the rollup tier + /// (`storedHourWeightSql`); the admin status route. pulseDayMeanMethod String @default("hour_mean_v1") @map("pulse_day_mean_method") - /// @pending-consumer(#v143-B8) When the admin applied the current method. + /// When the admin applied the current method; shown on the admin card. pulseDayMeanAppliedAt DateTime? @map("pulse_day_mean_applied_at") - /// @pending-consumer(#v143-B8) Last dry-run status and result (account labels only, no ids). + /// Last dry-run status and result (account ordinals only, no ids); the apply + /// step's precondition and the admin card's result. pulseDayMeanDryRunJson Json? @map("pulse_day_mean_dry_run_json") @@map("app_settings") diff --git a/src/__tests__/contract-stub-inventory.test.ts b/src/__tests__/contract-stub-inventory.test.ts index 3c6b7c5b1..293b57bf1 100644 --- a/src/__tests__/contract-stub-inventory.test.ts +++ b/src/__tests__/contract-stub-inventory.test.ts @@ -24,10 +24,6 @@ const API_ROOT = join(process.cwd(), "src", "app", "api"); /** Route file (under `src/app/api`) to the operations it still stubs. */ const CONTRACT_STUBS: Readonly> = { - "admin/maintenance/pulse-day-mean/route.ts": [ - "admin.pulse_day_mean.get", - "admin.pulse_day_mean.post", - ], "anamnesis/audit-c/[id]/route.ts": ["anamnesis.audit_c.delete"], "anamnesis/audit-c/route.ts": ["anamnesis.audit_c.create"], "anamnesis/life-circumstances/ai-sections/route.ts": [ @@ -53,7 +49,7 @@ const CONTRACT_STUBS: Readonly> = { }; /** The size of the register when it was written; see the header. */ -const REGISTER_SIZE = 15; +const REGISTER_SIZE = 14; function routeFiles(): string[] { return walkSourceFiles(API_ROOT, { floor: 400, extensions: [".ts"] }) diff --git a/src/__tests__/dense-measurement-read-guard.test.ts b/src/__tests__/dense-measurement-read-guard.test.ts index fa81848b6..eb46135e0 100644 --- a/src/__tests__/dense-measurement-read-guard.test.ts +++ b/src/__tests__/dense-measurement-read-guard.test.ts @@ -127,6 +127,8 @@ const ALLOWED_SQL: Record = { "Canonical-source blood pressure and weight over 90 days: sparse types, a few rows a day.", "lib/measurements/reconcile-external-measurement.ts::1": "Row locks (`FOR UPDATE`) on an id list the caller already read; returns those ids only.", + "lib/jobs/pulse-day-mean.ts::1": + "Selects one row per account; `measurements` is read only inside the MIN / MAX scalar subqueries of each row.", "lib/jobs/pr-detection.ts::1": "Selects account ids; `measurements` is read inside `EXISTS`, one row per account at most.", "lib/rollups/measurement-coverage.ts::1": diff --git a/src/__tests__/error-level-discipline-guard.test.ts b/src/__tests__/error-level-discipline-guard.test.ts index bd1b96be9..027fe7e17 100644 --- a/src/__tests__/error-level-discipline-guard.test.ts +++ b/src/__tests__/error-level-discipline-guard.test.ts @@ -201,6 +201,11 @@ const ALLOWED: Readonly> = { reason: "A size probe for an admin card: null shows no figure, and the status read must not fail over it.", }, + "lib/jobs/pulse-day-mean.ts::isPulseDayMeanRunActive": { + count: 1, + reason: + "No pgboss schema on a web-only deployment, so no run can be active; the route answers 503 there before it asks.", + }, "lib/jobs/job-observer.ts::emitJobLine": { count: 1, reason: diff --git a/src/__tests__/pulse-day-mean-reader-guard.test.ts b/src/__tests__/pulse-day-mean-reader-guard.test.ts index df329bd84..a94277515 100644 --- a/src/__tests__/pulse-day-mean-reader-guard.test.ts +++ b/src/__tests__/pulse-day-mean-reader-guard.test.ts @@ -26,6 +26,11 @@ * imports the shared statistic, hands the type to a fold that does, or * is listed with the reason it never averages pulse readings. * + * A third matcher has no list: since the pulse day-mean method (v1.43) an + * hour weighs by the method, so no day may still be folded from its hours' + * means with every hour weighing one (`AVG(hour_mean)`, a sum over hours + * divided by their number). The weight lives in the helper alone. + * * Comments are stripped before matching, the matchers are whitespace-tolerant, * and the floors fail the guard if either matcher stops finding anything. The * planted cases run each matcher over an offender; dropping the type from the @@ -67,6 +72,15 @@ const TS_MEAN = const NAMES_PULSE = /\bPULSE\b/; +/** + * A day folded from its hours' means with every hour weighing one: SQL + * `AVG(hour_mean)` and an in-memory sum over hours divided by their number. + * Since v1.43 an hour weighs by the pulse day-mean method, which only the + * helper applies (`hourWeightSql`, `combineHourCells`), so none may remain. + */ +const UNWEIGHTED_HOURS = + /\bAVG\s*\(\s*(?:[a-z_]+\s*\.\s*)?"?hour_mean"?\s*\)|\/\s*hours\.(?:length|size)\b/g; + type Entry = /** Averages pulse and imports the shared statistic. */ | { helper: true } @@ -85,6 +99,8 @@ const SQL_READERS: Record = { "lib/doctor-report/dense-buckets.ts": HELPER, "lib/insights/comprehensive-aggregator.ts": HELPER, "lib/insights/comprehensive-generate.ts": HELPER, + // The dry run's hour means, weighed by both methods through the helper. + "lib/jobs/pulse-day-mean.ts": HELPER, "lib/measurements/daily-series-read.ts": HELPER, "lib/rollups/measurement-rollups.ts": HELPER, "lib/rollups/tiered-context.ts": HELPER, @@ -163,7 +179,11 @@ const TS_READERS: Record = { }, }; -function scan(): { sql: Map; ts: Map } { +function scan(): { + sql: Map; + ts: Map; + unweighted: string[]; +} { const files = walkSourceFiles(SRC, { floor: 1500 }).filter( (f) => !f.startsWith("generated/") && @@ -173,8 +193,11 @@ function scan(): { sql: Map; ts: Map } { ); const sql = new Map(); const ts = new Map(); + const unweighted: string[] = []; for (const f of files) { const text = stripComments(readFileSync(join(SRC, f), "utf8")); + if (UNWEIGHTED_HOURS.test(text)) unweighted.push(f); + UNWEIGHTED_HOURS.lastIndex = 0; const s = text.match(SQL_MEAN)?.length ?? 0; if (s > 0) sql.set(f, s); if (NAMES_PULSE.test(text)) { @@ -182,10 +205,10 @@ function scan(): { sql: Map; ts: Map } { if (t > 0) ts.set(f, t); } } - return { sql, ts }; + return { sql, ts, unweighted }; } -const { sql, ts } = scan(); +const { sql, ts, unweighted } = scan(); const source = (f: string) => readFileSync(join(SRC, f), "utf8"); function check(found: Map, listed: Record) { @@ -239,6 +262,22 @@ describe("pulse day-mean reader guard", () => { }); }); + it("leaves no day folded from equally weighted hours", () => { + expect(unweighted).toEqual([]); + }); + + it("catches a planted unweighted hour fold", () => { + expect(`AVG(h.hour_mean) AS day_mean`.match(UNWEIGHTED_HOURS)).toHaveLength( + 1, + ); + expect( + `hours.reduce((s, h) => s + h.total / h.count, 0) / hours.length`.match( + UNWEIGHTED_HOURS, + ), + ).toHaveLength(1); + expect(`SUM(h.hour_mean * w) / SUM(w)`.match(UNWEIGHTED_HOURS)).toBeNull(); + }); + it("catches a planted plain AVG and a planted fold", () => { expect( `SELECT AVG( m."value" ) FROM measurements m`.match(SQL_MEAN), diff --git a/src/__tests__/success-affordance-guard.test.ts b/src/__tests__/success-affordance-guard.test.ts index 72619b731..2ef7b29f8 100644 --- a/src/__tests__/success-affordance-guard.test.ts +++ b/src/__tests__/success-affordance-guard.test.ts @@ -107,6 +107,11 @@ const PINNED_AFFORDANCES: Record< "src/components/admin/measurement-maintenance-card.tsx": { "toast.success": 1, }, + // Fires only on a 202: the dry run was queued, or the method was applied. + // The dry run's outcome itself is the status line. + "src/components/admin/pulse-day-mean-card.tsx": { + "toast.success": 2, + }, "src/components/admin/invite-tokens-section.tsx": { "text-success": 2, "toast.success": 3, diff --git a/src/app/admin/[section]/renderer.tsx b/src/app/admin/[section]/renderer.tsx index 6579e9ca8..942d7a770 100644 --- a/src/app/admin/[section]/renderer.tsx +++ b/src/app/admin/[section]/renderer.tsx @@ -41,6 +41,7 @@ import { SupportReportCard } from "@/components/admin/support-report-card"; import { RevealWhenSettled } from "@/components/ui/reveal-when-settled"; import { Skeleton } from "@/components/ui/skeleton"; import { MeasurementMaintenanceCard } from "@/components/admin/measurement-maintenance-card"; +import { PulseDayMeanCard } from "@/components/admin/pulse-day-mean-card"; import { InviteTokensSection } from "@/components/admin/invite-tokens-section"; import { UserManagementSection } from "@/components/admin/user-management-section"; import type { AdminSectionSlug } from "@/components/admin/section-slugs"; @@ -75,6 +76,8 @@ export function AdminSectionRenderer({ {/* v1.42 — the table maintenance after the tombstone purge. */} + {/* v1.43 — the pulse day-mean method, dry run before apply. */} + ); diff --git a/src/app/api/admin/maintenance/pulse-day-mean/__tests__/route.test.ts b/src/app/api/admin/maintenance/pulse-day-mean/__tests__/route.test.ts new file mode 100644 index 000000000..3f05cb0a1 --- /dev/null +++ b/src/app/api/admin/maintenance/pulse-day-mean/__tests__/route.test.ts @@ -0,0 +1,278 @@ +/** + * `GET/POST /api/admin/maintenance/pulse-day-mean` — the pulse day-mean + * method's dry run and apply step. Admin only; apply needs a finished dry run + * from the last seven days, and nothing starts while a run is active. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; + +vi.mock("@/lib/db", () => ({ + prisma: { + $queryRaw: vi.fn(), + appSettings: { findUnique: vi.fn(), upsert: vi.fn() }, + }, +})); +vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); +vi.mock("@/lib/auth/audit", () => ({ + auditLog: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() })); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("next/headers", () => ({ + headers: vi.fn(async () => ({ get: () => null })), + cookies: vi.fn(async () => ({ + get: () => undefined, + set: () => {}, + delete: () => {}, + })), +})); + +import { GET, POST } from "../route"; +import { prisma } from "@/lib/db"; +import { auditLog } from "@/lib/auth/audit"; +import { getSession } from "@/lib/auth/session"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + activePulseDayMeanMethod, + setActivePulseDayMeanMethod, +} from "@/lib/measurements/day-statistic"; + +const ADMIN = { + session: { id: "s-1", expiresAt: new Date(Date.now() + 3_600_000) }, + user: { id: "admin-1", username: "admin", role: "ADMIN" as const }, +}; +const USER = { + session: { id: "s-2", expiresAt: new Date(Date.now() + 3_600_000) }, + user: { id: "user-1", username: "bob", role: "USER" as const }, +}; + +const URL = "http://localhost/api/admin/maintenance/pulse-day-mean"; +const get = () => + (GET as unknown as (r: NextRequest) => Promise)( + new NextRequest(URL), + ); +const post = (body: unknown) => + (POST as unknown as (r: NextRequest) => Promise)( + new NextRequest(URL, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }), + ); + +const NOW = new Date("2026-10-16T12:00:00Z"); +const send = vi.fn(); + +function settings(dryRun: unknown, method = "hour_mean_v1") { + vi.mocked(prisma.appSettings.findUnique).mockResolvedValue({ + pulseDayMeanMethod: method, + pulseDayMeanAppliedAt: null, + pulseDayMeanDryRunJson: dryRun, + } as never); +} + +function activeJobs(n: number) { + vi.mocked(prisma.$queryRaw).mockResolvedValue([{ n }] as never); +} + +const RESULT = { + accounts: 1, + daysCompared: 30, + daysChangedOver1Bpm: 2, + maxChangeBpm: 3.4, + distribution: [ + { bucket: "0-1", days: 28 }, + { bucket: "1-2", days: 1 }, + { bucket: "2-5", days: 1 }, + { bucket: "5+", days: 0 }, + ], + perAccount: [{ account: "#1", days: 30, changedOver1Bpm: 2 }], +}; + +function doneAt(iso: string) { + return { + status: "done", + startedAt: iso, + finishedAt: iso, + result: RESULT, + }; +} + +beforeEach(() => { + vi.resetAllMocks(); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(NOW); + setActivePulseDayMeanMethod("hour_mean_v1"); + vi.mocked(getSession).mockResolvedValue(ADMIN as never); + vi.mocked(getGlobalBoss).mockReturnValue({ send } as never); + vi.mocked(prisma.appSettings.upsert).mockResolvedValue({} as never); + send.mockResolvedValue("job-1"); + activeJobs(0); +}); + +afterEach(() => { + vi.useRealTimers(); + setActivePulseDayMeanMethod("hour_mean_v1"); +}); + +describe("admin only", () => { + it("refuses anyone but an admin, before reading anything", async () => { + vi.mocked(getSession).mockResolvedValue(USER as never); + expect((await get()).status).toBe(403); + expect((await post({ action: "dry_run" })).status).toBe(403); + expect(prisma.appSettings.findUnique).not.toHaveBeenCalled(); + expect(send).not.toHaveBeenCalled(); + }); +}); + +describe("GET", () => { + it("answers the method and the last dry run", async () => { + settings(doneAt("2026-10-15T08:00:00.000Z")); + const body = await (await get()).json(); + expect(body.data).toEqual({ + method: "hour_mean_v1", + appliedAt: null, + lastDryRun: doneAt("2026-10-15T08:00:00.000Z"), + }); + }); + + it("reads a run marked running with no live job behind it as failed", async () => { + settings({ + status: "running", + startedAt: "2026-10-16T11:00:00.000Z", + finishedAt: null, + result: null, + }); + activeJobs(0); + const body = await (await get()).json(); + expect(body.data.lastDryRun.status).toBe("failed"); + }); + + it("drops a stored value of the wrong shape instead of passing it on", async () => { + settings({ status: "done", userId: "u-1" }); + const body = await (await get()).json(); + expect(body.data.lastDryRun).toBeNull(); + }); +}); + +describe("POST dry_run", () => { + it("marks the dry run queued and enqueues it", async () => { + settings(null); + const res = await post({ action: "dry_run" }); + expect(res.status).toBe(202); + expect((await res.json()).data).toEqual({ enqueued: true }); + expect(send).toHaveBeenCalledWith( + "pulse-day-mean", + { action: "dry_run" }, + expect.objectContaining({ + retryLimit: 0, + singletonKey: "pulse-day-mean", + }), + ); + expect(prisma.appSettings.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + update: { + pulseDayMeanDryRunJson: expect.objectContaining({ status: "queued" }), + }, + }), + ); + }); + + it("answers 409 busy while a run is active, and starts nothing", async () => { + activeJobs(1); + const res = await post({ action: "dry_run" }); + expect(res.status).toBe(409); + expect((await res.json()).meta.errorCode).toBe( + "maintenance.pulseDayMean.busy", + ); + expect(send).not.toHaveBeenCalled(); + expect(prisma.appSettings.upsert).not.toHaveBeenCalled(); + }); + + it("puts the last dry run back when the send loses the singleton", async () => { + const previous = doneAt("2026-10-15T08:00:00.000Z"); + settings(previous); + send.mockResolvedValue(null); + const res = await post({ action: "dry_run" }); + expect(res.status).toBe(409); + expect(vi.mocked(prisma.appSettings.upsert).mock.lastCall?.[0]).toEqual( + expect.objectContaining({ update: { pulseDayMeanDryRunJson: previous } }), + ); + }); +}); + +describe("POST apply", () => { + it("is refused without a dry run", async () => { + settings(null); + const res = await post({ action: "apply" }); + expect(res.status).toBe(409); + expect((await res.json()).meta.errorCode).toBe( + "maintenance.pulseDayMean.dryRunRequired", + ); + expect(prisma.appSettings.upsert).not.toHaveBeenCalled(); + expect(send).not.toHaveBeenCalled(); + }); + + it("is refused with a dry run older than seven days", async () => { + settings(doneAt("2026-10-09T11:59:00.000Z")); + expect((await post({ action: "apply" })).status).toBe(409); + }); + + it("is refused with a dry run that failed", async () => { + settings({ ...doneAt("2026-10-15T08:00:00.000Z"), status: "failed" }); + expect((await post({ action: "apply" })).status).toBe(409); + }); + + it("is refused 409 busy while a run is active", async () => { + settings(doneAt("2026-10-15T08:00:00.000Z")); + activeJobs(1); + const res = await post({ action: "apply" }); + expect((await res.json()).meta.errorCode).toBe( + "maintenance.pulseDayMean.busy", + ); + }); + + it("switches the method, enqueues the rebuild and audits it", async () => { + settings(doneAt("2026-10-15T08:00:00.000Z")); + const res = await post({ action: "apply" }); + expect(res.status).toBe(202); + expect((await res.json()).data).toEqual({ enqueued: true }); + expect(prisma.appSettings.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + update: { + pulseDayMeanMethod: "weighted_hour_mean_v2", + pulseDayMeanAppliedAt: NOW, + }, + }), + ); + expect(send).toHaveBeenCalledWith( + "pulse-day-mean", + { action: "rebuild" }, + expect.anything(), + ); + expect(activePulseDayMeanMethod()).toBe("weighted_hour_mean_v2"); + expect(auditLog).toHaveBeenCalledWith( + "admin.pulseDayMean.apply", + expect.objectContaining({ + userId: "admin-1", + details: { method: "weighted_hour_mean_v2", rebuildEnqueued: true }, + }), + ); + }); +}); + +describe("POST body", () => { + it("is strict", async () => { + expect((await post({ action: "apply", force: true })).status).toBe(422); + expect((await post({ action: "rebuild" })).status).toBe(422); + expect(send).not.toHaveBeenCalled(); + }); + + it("answers 503 without a background worker", async () => { + vi.mocked(getGlobalBoss).mockReturnValue(null as never); + expect((await post({ action: "dry_run" })).status).toBe(503); + }); +}); diff --git a/src/app/api/admin/maintenance/pulse-day-mean/route.ts b/src/app/api/admin/maintenance/pulse-day-mean/route.ts index f5bb44f50..c94e34f50 100644 --- a/src/app/api/admin/maintenance/pulse-day-mean/route.ts +++ b/src/app/api/admin/maintenance/pulse-day-mean/route.ts @@ -3,30 +3,72 @@ * 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 { pulseDayMeanActionSchema } from "@/lib/validations/pulse-day-mean"; +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 { + 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 { + hasFreshDryRun, + 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 +76,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); }); diff --git a/src/app/api/measurements/series/route.ts b/src/app/api/measurements/series/route.ts index 72901ddb9..b3abf5504 100644 --- a/src/app/api/measurements/series/route.ts +++ b/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 diff --git a/src/components/admin/__tests__/pulse-day-mean-card.test.tsx b/src/components/admin/__tests__/pulse-day-mean-card.test.tsx new file mode 100644 index 000000000..926de7ea6 --- /dev/null +++ b/src/components/admin/__tests__/pulse-day-mean-card.test.tsx @@ -0,0 +1,116 @@ +/** + * The pulse day-mean admin card. + * + * 1. apply is offered only after a dry run finished in the last seven days, + * and never while a dry run is queued or running; + * 2. a finished dry run shows its figures and the per-account ordinals; + * 3. the active method reads in the header, in German too. + * + * Mutation check: drop `hasFreshDryRun` from `canApply` and case 1 goes red. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { renderToStaticMarkup } from "react-dom/server"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; + +import { PulseDayMeanCard } from "@/components/admin/pulse-day-mean-card"; +import { I18nProvider } from "@/lib/i18n/context"; +import { queryKeys } from "@/lib/query-keys"; +import type { PulseDayMeanStatus } from "@/lib/validations/pulse-day-mean"; + +const RESULT = { + accounts: 2, + daysCompared: 412, + daysChangedOver1Bpm: 37, + maxChangeBpm: 6.4, + distribution: [ + { bucket: "0-1" as const, days: 375 }, + { bucket: "1-2" as const, days: 30 }, + { bucket: "2-5" as const, days: 6 }, + { bucket: "5+" as const, days: 1 }, + ], + perAccount: [ + { account: "#1", days: 400, changedOver1Bpm: 36 }, + { account: "#2", days: 12, changedOver1Bpm: 1 }, + ], +}; + +const status = ( + lastDryRun: PulseDayMeanStatus["lastDryRun"], + method: PulseDayMeanStatus["method"] = "hour_mean_v1", +): PulseDayMeanStatus => ({ method, appliedAt: null, lastDryRun }); + +const done = (finishedAt: string) => ({ + status: "done" as const, + startedAt: finishedAt, + finishedAt, + result: RESULT, +}); + +function render(s: PulseDayMeanStatus, locale: "en" | "de" = "en"): string { + const client = new QueryClient(); + client.setQueryData(queryKeys.adminPulseDayMean(), s); + return renderToStaticMarkup( + + + + + , + ); +} + +/** Whether the apply button renders disabled. */ +function applyDisabled(html: string): boolean { + const at = html.indexOf('data-testid="admin-pulse-day-mean-apply"'); + const tag = html.slice( + html.lastIndexOf("", at), + ); + return /\sdisabled=""/.test(tag); +} + +beforeEach(() => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-10-16T12:00:00Z")); +}); +afterEach(() => vi.useRealTimers()); + +describe("PulseDayMeanCard", () => { + it("offers apply only after a fresh, finished dry run", () => { + expect(applyDisabled(render(status(null)))).toBe(true); + expect( + applyDisabled(render(status(done("2026-10-08T12:00:00.000Z")))), + ).toBe(true); + expect( + applyDisabled( + render( + status({ + status: "running", + startedAt: "2026-10-16T11:00:00.000Z", + finishedAt: null, + result: null, + }), + ), + ), + ).toBe(true); + expect( + applyDisabled(render(status(done("2026-10-15T12:00:00.000Z")))), + ).toBe(false); + }); + + it("shows what the dry run found", () => { + const html = render(status(done("2026-10-15T12:00:00.000Z"))); + expect(html).toContain("412"); + expect(html).toContain("37"); + expect(html).toContain("6.4 bpm"); + expect(html).toContain("5+ bpm"); + expect(html).toContain("Account #2"); + expect(html).toContain("36 of 400 days move"); + }); + + it("names the active method, in German too", () => { + expect(render(status(null))).toContain("Every hour equal"); + const de = render(status(null, "weighted_hour_mean_v2"), "de"); + expect(de).toContain("Dünne Stunden gewichtet"); + expect(de).toContain("Probelauf"); + }); +}); diff --git a/src/components/admin/pulse-day-mean-card.tsx b/src/components/admin/pulse-day-mean-card.tsx new file mode 100644 index 000000000..285fec109 --- /dev/null +++ b/src/components/admin/pulse-day-mean-card.tsx @@ -0,0 +1,306 @@ +"use client"; + +/** + * `` — switch the pulse day mean to weigh sparse hours, with + * a dry run first (#1107 follow-up). + * + * The card shows the active method, starts a dry run, shows what it found + * (how many days would move by more than 1 bpm, the largest move, the spread, + * and the same per account under an ordinal), and applies the new method once + * a dry run from the last seven days has finished. It polls only while a dry + * run is queued or running. The route enforces every rule the buttons mirror. + */ +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { FlaskConical, HeartPulse, Loader2 } from "lucide-react"; +import { toast } from "sonner"; + +import { StatTile } from "@/components/admin/_stat-tile"; +import { SettingsCardActions } from "@/components/settings/_card-actions"; +import { SettingsCardHeader } from "@/components/settings/_card-header"; +import { SettingsCard } from "@/components/settings/settings-card"; +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, + AlertDialogTrigger, +} from "@/components/ui/alert-dialog"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { QueryErrorCard } from "@/components/ui/query-error-card"; +import { apiFetchRaw, apiGet } from "@/lib/api/api-fetch"; +import { showErrorToast } from "@/lib/errors/error-toast"; +import { useFormatters, useTranslations } from "@/lib/i18n/context"; +import { queryKeys } from "@/lib/query-keys"; +import { + hasFreshDryRun, + type PulseDayMeanActionInput, + type PulseDayMeanStatus, +} from "@/lib/validations/pulse-day-mean"; + +const ENDPOINT = "/api/admin/maintenance/pulse-day-mean"; + +/** How often the card asks while a dry run is waiting or working. */ +const POLL_MS = 10_000; + +export function PulseDayMeanCard() { + const { t } = useTranslations(); + const fmt = useFormatters(); + const queryClient = useQueryClient(); + + const status = useQuery({ + queryKey: queryKeys.adminPulseDayMean(), + queryFn: () => + apiGet(ENDPOINT, { credentials: "include" }), + refetchInterval: (query) => { + const state = query.state.data?.lastDryRun?.status; + return state === "queued" || state === "running" ? POLL_MS : false; + }, + }); + + const run = useMutation({ + mutationFn: async ( + action: PulseDayMeanActionInput["action"], + ): Promise<{ action: typeof action; enqueued: boolean }> => { + const res = await apiFetchRaw(ENDPOINT, { + method: "POST", + credentials: "include", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ action }), + }); + if (!res.ok) { + let code: string | undefined; + try { + const body = (await res.json()) as { meta?: { errorCode?: string } }; + code = body.meta?.errorCode; + } catch { + // No envelope: the generic message below. + } + // The route's two refusals each have their own message. + throw new Error( + code === "maintenance.pulseDayMean.busy" + ? t("admin.section.pulseDayMean.error.busy") + : code === "maintenance.pulseDayMean.dryRunRequired" + ? t("admin.section.pulseDayMean.error.dryRunRequired") + : t("admin.section.pulseDayMean.error.generic"), + ); + } + const { enqueued } = ( + (await res.json()) as { data: { enqueued: boolean } } + ).data; + return { action, enqueued }; + }, + onSuccess: ({ action, enqueued }) => { + if (action === "apply") { + toast.success( + enqueued + ? t("admin.section.pulseDayMean.applied") + : t("admin.section.pulseDayMean.appliedNoRebuild"), + ); + } else { + toast.success(t("admin.section.pulseDayMean.dryRunStarted")); + } + void queryClient.invalidateQueries({ + queryKey: queryKeys.adminPulseDayMean(), + }); + }, + onError: (err: Error) => { + showErrorToast(err.message, { cause: err }); + }, + }); + + if (status.isError) { + return ( + void status.refetch()} + /> + ); + } + + const s = status.data; + const dryRun = s?.lastDryRun ?? null; + const result = dryRun?.status === "done" ? dryRun.result : null; + const active = dryRun?.status === "queued" || dryRun?.status === "running"; + const weighted = s?.method === "weighted_hour_mean_v2"; + const canApply = !!s && !active && hasFreshDryRun(s); + + return ( + + + {weighted + ? t("admin.section.pulseDayMean.method.weighted") + : t("admin.section.pulseDayMean.method.hourly")} + + ) : null + } + /> +

{t("admin.section.pulseDayMean.body")}

+ +
+ {status.isLoading ? ( + + + {t("admin.section.pulseDayMean.loading")} + + ) : active ? ( + + + {dryRun?.status === "queued" + ? t("admin.section.pulseDayMean.state.queued") + : t("admin.section.pulseDayMean.state.running")} + + ) : dryRun?.status === "done" ? ( + + {t("admin.section.pulseDayMean.state.done", { + at: fmt.dateTime(dryRun.finishedAt ?? dryRun.startedAt), + })} + + ) : dryRun?.status === "failed" ? ( + + {t("admin.section.pulseDayMean.state.failed")} + + ) : ( + + {t("admin.section.pulseDayMean.state.idle")} + + )} +
+ + {result ? ( +
+
+ + + +
+
+ {result.distribution.map((d) => ( +
+
+ {t("admin.section.pulseDayMean.result.bucket", { + range: d.bucket, + })} +
+
{fmt.number(d.days)}
+
+ ))} +
+ {result.perAccount.length > 0 ? ( +
    + {result.perAccount.map((a) => ( +
  • + + {t("admin.section.pulseDayMean.result.account", { + account: a.account, + })} + + + {t("admin.section.pulseDayMean.result.accountDays", { + changed: fmt.number(a.changedOver1Bpm), + days: fmt.number(a.days), + })} + +
    • + ))} +
+ ) : null} +
+ ) : null} + + {s?.appliedAt ? ( +

+ {t("admin.section.pulseDayMean.appliedAt", { + at: fmt.dateTime(s.appliedAt), + })} +

+ ) : null} + + + + + + + + + + + {t("admin.section.pulseDayMean.confirmTitle")} + + + {t("admin.section.pulseDayMean.confirmBody")} + + + + + {t("admin.section.pulseDayMean.cancel")} + + run.mutate("apply")}> + {t("admin.section.pulseDayMean.confirm")} + + + + + + + ); +} diff --git a/src/instrumentation.ts b/src/instrumentation.ts index 78981c5e6..7927effd9 100644 --- a/src/instrumentation.ts +++ b/src/instrumentation.ts @@ -39,6 +39,16 @@ export async function register() { // A diagnostic must never break boot. } + // Every pulse reader follows the operator's pulse day-mean method; mirror + // it in this process and keep the mirror current. + try { + const { startPulseDayMeanMethodSync } = + await import("@/lib/measurements/pulse-day-mean-method"); + await startPulseDayMeanMethodSync(); + } catch { + // Without the mirror the process keeps the default method. + } + // Every Node process watches its own event loop — web-only // containers included; a stall report from the process serving // requests is the whole point. diff --git a/src/lib/day/daily-stats.ts b/src/lib/day/daily-stats.ts index a9bcba033..d3fa3692f 100644 --- a/src/lib/day/daily-stats.ts +++ b/src/lib/day/daily-stats.ts @@ -23,6 +23,7 @@ import { } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; import { isCumulativeDaySumType } from "@/lib/measurements/cumulative-day-sum"; +import { combineHourCells } from "@/lib/measurements/day-mean"; import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; import { dateOnlyAtNoonUtc, dateOnlyKey } from "@/lib/tz/date-only"; import { collapseRollupRowsBySource } from "@/lib/rollups/measurement-read"; @@ -169,8 +170,9 @@ export function foldDailyCells( if (cumulative) { value = hours.reduce((s, h) => s + h.total, 0); } else if (hourly) { - value = - hours.reduce((s, h) => s + h.total / h.count, 0) / hours.length; + value = combineHourCells( + hours.map((h) => ({ sum: h.total, n: h.count })), + ) as number; } else { value = hours.reduce((s, h) => s + h.total, 0) / count; } diff --git a/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts index b83ac5e05..54dc53f7a 100644 --- a/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts +++ b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts @@ -45,8 +45,8 @@ function bucket( clinicalTar2Count: 0, clinicalLowRiskSum: null, clinicalHighRiskSum: null, - hourMeanSum: null, - hourCount: null, + hourWeightedSum: null, + hourWeight: null, }; } diff --git a/src/lib/doctor-report/dense-buckets.ts b/src/lib/doctor-report/dense-buckets.ts index 8d05fdab9..3810fcda7 100644 --- a/src/lib/doctor-report/dense-buckets.ts +++ b/src/lib/doctor-report/dense-buckets.ts @@ -13,7 +13,10 @@ */ import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; -import { isHourlyMeanTypeSql } from "@/lib/measurements/day-mean"; +import { + hourWeightSql, + isHourlyMeanTypeSql, +} from "@/lib/measurements/day-mean"; import type { GlucoseContext, MeasurementSource, @@ -47,11 +50,12 @@ export interface DenseMeasurementBucket { clinicalHighRiskSum: number | null; /** * For an hourly-mean type (pulse, see `day-mean.ts`): the sum of the - * bucket's local hours' means and the number of those hours, so the day - * value is their ratio. `null` for every other type. + * bucket's local hours' means, each times its hour weight, and the sum of + * those weights, so the day value is their ratio. `null` for every other + * type. */ - hourMeanSum: number | null; - hourCount: number | null; + hourWeightedSum: number | null; + hourWeight: number | null; } /** @@ -75,6 +79,8 @@ export async function loadDenseMeasurementBuckets(params: { ); // Closed code constant (the hourly-mean type list), spliced as literals. const hourlyMeanType = Prisma.raw(isHourlyMeanTypeSql('"type"')); + // Each hour's weight under the active pulse day-mean method (code-built). + const hourWeight = Prisma.raw(hourWeightSql("n")); return prisma.$queryRaw` WITH localized AS ( SELECT @@ -135,8 +141,9 @@ export async function loadDenseMeasurementBuckets(params: { "measured_at" DESC, "id" DESC ), - -- The hourly-mean day (pulse): each local hour's mean, then their sum - -- and count per bucket. Empty for every other type. + -- The hourly-mean day (pulse): each local hour's mean and reading + -- count, then the weighted sum of the means and the sum of the hour + -- weights per bucket. Empty for every other type. hour_means AS ( SELECT local_day, @@ -144,7 +151,8 @@ export async function loadDenseMeasurementBuckets(params: { "source", "device_type", "glucose_context", - AVG("value") AS hour_mean + AVG("value") AS hour_mean, + COUNT(*) AS n FROM localized WHERE ${hourlyMeanType} GROUP BY @@ -162,8 +170,8 @@ export async function loadDenseMeasurementBuckets(params: { "source", "device_type", "glucose_context", - SUM(hour_mean)::double precision AS hour_mean_sum, - COUNT(*)::int AS hour_count + SUM(hour_mean * ${hourWeight})::double precision AS hour_weighted_sum, + SUM(${hourWeight})::double precision AS hour_weight FROM hour_means GROUP BY local_day, @@ -248,8 +256,8 @@ export async function loadDenseMeasurementBuckets(params: { grouped.clinical_tar2_count AS "clinicalTar2Count", grouped.clinical_low_risk_sum AS "clinicalLowRiskSum", grouped.clinical_high_risk_sum AS "clinicalHighRiskSum", - hour_agg.hour_mean_sum AS "hourMeanSum", - hour_agg.hour_count AS "hourCount" + hour_agg.hour_weighted_sum AS "hourWeightedSum", + hour_agg.hour_weight AS "hourWeight" FROM grouped INNER JOIN latest_rows ON latest_rows.local_day = grouped.local_day diff --git a/src/lib/doctor-report/measurement-series.ts b/src/lib/doctor-report/measurement-series.ts index c00d76eb0..e353f3388 100644 --- a/src/lib/doctor-report/measurement-series.ts +++ b/src/lib/doctor-report/measurement-series.ts @@ -62,8 +62,8 @@ export function summariseDenseBuckets( max: number; latestAt: Date; latest: number; - hourMeanSum: number; - hourCount: number; + hourWeightedSum: number; + hourWeight: number; } >(); for (const row of canonical) { @@ -79,15 +79,15 @@ export function summariseDenseBuckets( max: row.maxValue, latestAt: row.latestAt, latest: row.latestValue, - hourMeanSum: row.hourMeanSum ?? 0, - hourCount: row.hourCount ?? 0, + hourWeightedSum: row.hourWeightedSum ?? 0, + hourWeight: row.hourWeight ?? 0, }); continue; } current.count += row.count; current.sum += row.sumValue; - current.hourMeanSum += row.hourMeanSum ?? 0; - current.hourCount += row.hourCount ?? 0; + current.hourWeightedSum += row.hourWeightedSum ?? 0; + current.hourWeight += row.hourWeight ?? 0; current.min = Math.min(current.min, row.minValue); current.max = Math.max(current.max, row.maxValue); if (row.latestAt > current.latestAt) { @@ -116,16 +116,16 @@ export function summariseDenseBuckets( } }; - // For an hourly-mean type (pulse, see `day-mean.ts`) a day is the mean of - // its local hours' means and the period average the mean of its days, so a - // workout hour's dense samples do not outweigh the rest of the day, nor a - // workout day the rest of the period. Count, min and max stay over every + // For an hourly-mean type (pulse, see `day-mean.ts`) a day is the weighted + // mean of its local hours' means and the period average the mean of its + // days, so a workout hour's dense samples do not outweigh the rest of the + // day, nor a workout day the rest of the period. Count, min and max stay over every // reading. Every other type keeps the mean of its readings. const dayMeans = new Map(); for (const row of byTypeDay.values()) { const value = - usesHourlyMeanDay(row.type) && row.hourCount > 0 - ? row.hourMeanSum / row.hourCount + usesHourlyMeanDay(row.type) && row.hourWeight > 0 + ? row.hourWeightedSum / row.hourWeight : row.sum / row.count; (byType[row.type] ??= []).push({ value, diff --git a/src/lib/insights/correlation-channel-series.ts b/src/lib/insights/correlation-channel-series.ts index a0f6d8fce..0d1db8cf8 100644 --- a/src/lib/insights/correlation-channel-series.ts +++ b/src/lib/insights/correlation-channel-series.ts @@ -67,7 +67,7 @@ import type { } from "@/generated/prisma/client"; import { TRACKED_INTAKE_WHERE } from "@/lib/medications/intake-tracking"; import { dateOnlyKey } from "@/lib/tz/date-only"; -import { hourMeanRows } from "@/lib/measurements/day-mean"; +import { dayValueRows } from "@/lib/measurements/day-mean"; import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; import { decryptSymptomText } from "@/lib/symptoms/server"; import { @@ -750,11 +750,11 @@ export function buildMeasurementDailySeries( if (CUMULATIVE_HK_TYPES.has(type)) { return buildCumulativeDailySeries(type, rows, tz, priorityJson); } - // A day of pulse is the mean of its local hours' means (`day-mean.ts`): - // collapse each local hour to its mean first, so the daily mean below - // weighs every hour once. Every other type passes through unchanged. + // A day of pulse is the weighted mean of its local hours' means + // (`day-mean.ts`): collapse each local day to that value first, so the daily + // mean below passes it through. Every other type passes through unchanged. return toDailyMeans( - hourMeanRows( + dayValueRows( type, rows.map((r) => ({ value: r.value, measuredAt: r.at })), tz, diff --git a/src/lib/jobs/pulse-day-mean.ts b/src/lib/jobs/pulse-day-mean.ts new file mode 100644 index 000000000..f90d4a3a6 --- /dev/null +++ b/src/lib/jobs/pulse-day-mean.ts @@ -0,0 +1,430 @@ +/** + * The pulse day-mean method switch (#1107 follow-up): dry run, then apply. + * + * Both start from `POST /api/admin/maintenance/pulse-day-mean` (cookie-only + * admin) and run here, in the background worker, one at a time. + * + * - `dry_run` computes, for every account with pulse and every UTC day, the + * day value under both methods on the account's canonical source, and + * writes one thing: the result, to `app_settings.pulse_day_mean_dry_run_json` + * (how many days move by more than 1 bpm, the largest move, a distribution, + * and the same per account under an ordinal label). Nothing else is + * written, the rollup tier included. + * - `rebuild` follows the apply step, which has already stored the new + * method: it recomputes the PULSE rollups of every account, which read the + * stored method inside their own statement. + * + * Both work one account and one window at a time (`DRY_RUN_WINDOW_DAYS`, + * `REBUILD_WINDOW_DAYS`), so the process holds a window's aggregate rows and + * never an account's readings: the heap stays flat however long the history. + */ +import type { Job } from "pg-boss"; + +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; +import { prisma } from "@/lib/db"; +import { withJobLock } from "@/lib/jobs/job-lock"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { logCaught } from "@/lib/logging/signal"; +import { + hourWeightSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; +import { + isPulseDayMeanMethod, + DEFAULT_PULSE_DAY_MEAN_METHOD, +} from "@/lib/measurements/day-statistic"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { + recomputeUserRollups, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; +import { + PULSE_DAY_MEAN_CHANGE_BUCKETS, + pulseDayMeanDryRunSchema, + type PulseDayMeanDryRun, + type PulseDayMeanDryRunResult, + type PulseDayMeanStatus, +} from "@/lib/validations/pulse-day-mean"; + +export const PULSE_DAY_MEAN_QUEUE = "pulse-day-mean"; + +/** + * One-shot and exclusive: a dry run or a rebuild that stops part way is + * started again by the operator, never retried beside a half-finished one. + */ +export const PULSE_DAY_MEAN_SEND_OPTIONS = { + retryLimit: 0, + expireInSeconds: 6 * 60 * 60, + singletonKey: PULSE_DAY_MEAN_QUEUE, +} as const; + +export interface PulseDayMeanPayload { + action: "dry_run" | "rebuild"; +} + +const DAY_MS = 24 * 60 * 60 * 1000; +/** Days of one account's pulse a dry-run statement covers. */ +const DRY_RUN_WINDOW_DAYS = 90; +/** Days of one account's pulse a rebuild statement covers. */ +const REBUILD_WINDOW_DAYS = 366; +/** A window's statements may outlast the request limit, never this. */ +const STATEMENT_BUDGET_MS = 5 * 60 * 1000; + +/** A day "changed" when its value moves by more than this many bpm. */ +const CHANGED_OVER_BPM = 1; + +// ─── Status ─────────────────────────────────────────────── + +/** Whether a dry run or rebuild is queued or running. */ +export async function isPulseDayMeanRunActive(): Promise { + try { + const rows = await prisma.$queryRaw>` + SELECT count(*)::int AS n + FROM pgboss.job + WHERE name = ${PULSE_DAY_MEAN_QUEUE} + AND state IN ('created', 'retry', 'active') + `; + return (rows[0]?.n ?? 0) > 0; + } catch { + // No queue schema (web-only deployment): nothing can be running. + return false; + } +} + +function parseDryRun(value: unknown): PulseDayMeanDryRun | null { + const parsed = pulseDayMeanDryRunSchema.safeParse(value); + return parsed.success ? parsed.data : null; +} + +/** + * The stored method, when it was applied, and the last dry run. A dry run + * still marked queued or running with no live job behind it (the worker went + * away mid-run) reads as failed. + */ +export async function readPulseDayMeanStatus(): Promise { + const row = await prisma.appSettings.findUnique({ + where: { id: "singleton" }, + select: { + pulseDayMeanMethod: true, + pulseDayMeanAppliedAt: true, + pulseDayMeanDryRunJson: true, + }, + }); + let lastDryRun = parseDryRun(row?.pulseDayMeanDryRunJson ?? null); + if ( + lastDryRun && + (lastDryRun.status === "queued" || lastDryRun.status === "running") && + !(await isPulseDayMeanRunActive()) + ) { + lastDryRun = { ...lastDryRun, status: "failed" }; + } + return { + method: isPulseDayMeanMethod(row?.pulseDayMeanMethod) + ? row.pulseDayMeanMethod + : DEFAULT_PULSE_DAY_MEAN_METHOD, + appliedAt: row?.pulseDayMeanAppliedAt?.toISOString() ?? null, + lastDryRun, + }; +} + +/** The only write a dry run makes. */ +export async function writeDryRun(run: PulseDayMeanDryRun): Promise { + await prisma.appSettings.upsert({ + where: { id: "singleton" }, + create: { id: "singleton", pulseDayMeanDryRunJson: run }, + update: { pulseDayMeanDryRunJson: run }, + }); +} + +// ─── Accounts ───────────────────────────────────────────── + +interface PulseAccount { + id: string; + first: Date; + last: Date; +} + +/** Accounts holding pulse, oldest account first, with their pulse span. */ +async function pulseAccounts(): Promise { + const rows = await prisma.$queryRaw< + Array<{ id: string; first: Date | null; last: Date | null }> + >` + SELECT u."id", + (SELECT MIN(m."measured_at") FROM measurements m + WHERE m."user_id" = u."id" AND m."type" = 'PULSE' + AND m."deleted_at" IS NULL) AS first, + (SELECT MAX(m."measured_at") FROM measurements m + WHERE m."user_id" = u."id" AND m."type" = 'PULSE' + AND m."deleted_at" IS NULL) AS last + FROM users u + ORDER BY u."created_at", u."id" + `; + return rows + .filter((r) => r.first !== null && r.last !== null) + .map((r) => ({ id: r.id, first: r.first!, last: r.last! })); +} + +/** `[from, to)` windows of `days` covering `[first, last]`, UTC-day aligned. */ +function windows( + first: Date, + last: Date, + days: number, +): Array<{ from: Date; to: Date }> { + const out: Array<{ from: Date; to: Date }> = []; + 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)), + }); + } + return out; +} + +// ─── Dry run ────────────────────────────────────────────── + +interface WindowComparison { + days: number; + over1: number; + maxDelta: number; + buckets: [number, number, number, number]; +} + +/** + * One account's UTC days in `[from, to)`: each day's value on its canonical + * source under both methods, compared. Aggregated in the statement; one row + * comes back however many readings the window holds. + */ +async function compareWindow( + userId: string, + rank: string, + from: Date, + to: Date, +): Promise { + const v1 = weightedHourMeanSql( + "h.hour_mean", + hourWeightSql("h.n", "hour_mean_v1"), + ); + const v2 = weightedHourMeanSql( + "h.hour_mean", + hourWeightSql("h.n", "weighted_hour_mean_v2"), + ); + // Whitelist splice: `v1` / `v2` are built from code constants and `rank` + // from the closed source enum (`buildSourceRankCase`); every value is a + // positional parameter. + const rows = await prisma.$transaction(async (tx) => { + await tx.$executeRaw`SELECT set_config('statement_timeout', ${String(STATEMENT_BUDGET_MS)}, true)`; + return tx.$queryRawUnsafe< + Array<{ + days: number; + over1: number; + max_delta: number; + b0: number; + b1: number; + b2: number; + b3: number; + }> + >( + ` + WITH h AS ( + SELECT m."source", + date_trunc('hour', m."measured_at") AS hr, + AVG(m."value") AS hour_mean, + COUNT(*) AS n + FROM measurements m + WHERE m."user_id" = $1 + AND m."type" = 'PULSE' + AND m."deleted_at" IS NULL + AND m."measured_at" >= $2 + AND m."measured_at" < $3 + GROUP BY 1, 2 + ), + d AS ( + SELECT h."source", date_trunc('day', h.hr) AS day, + ${v1} AS v1, + ${v2} AS v2 + FROM h + GROUP BY 1, 2 + ), + c AS ( + SELECT DISTINCT ON (d.day) ABS(d.v2 - d.v1) AS delta + FROM d + ORDER BY d.day, (${rank}), d."source"::text + ) + SELECT COUNT(*)::int AS days, + COUNT(*) FILTER (WHERE delta > $4)::int AS over1, + COALESCE(MAX(delta), 0)::double precision AS max_delta, + COUNT(*) FILTER (WHERE delta <= 1)::int AS b0, + COUNT(*) FILTER (WHERE delta > 1 AND delta <= 2)::int AS b1, + COUNT(*) FILTER (WHERE delta > 2 AND delta <= 5)::int AS b2, + COUNT(*) FILTER (WHERE delta > 5)::int AS b3 + FROM c + `, + userId, + from, + to, + CHANGED_OVER_BPM, + ); + }); + const r = rows[0]; + return { + days: Number(r?.days ?? 0), + over1: Number(r?.over1 ?? 0), + maxDelta: Number(r?.max_delta ?? 0), + buckets: [ + Number(r?.b0 ?? 0), + Number(r?.b1 ?? 0), + Number(r?.b2 ?? 0), + Number(r?.b3 ?? 0), + ], + }; +} + +/** Compare both methods over every account's pulse. Writes nothing. */ +export async function computePulseDayMeanDryRun(): Promise { + const buckets = [0, 0, 0, 0]; + let daysCompared = 0; + let over1 = 0; + let maxChange = 0; + const perAccount: PulseDayMeanDryRunResult["perAccount"] = []; + const accounts = await pulseAccounts(); + for (const [i, account] of accounts.entries()) { + const rank = buildSourceRankCase( + await loadUserSourcePriority(account.id), + "'PULSE'", + 'd."source"', + ); + let days = 0; + let changed = 0; + for (const w of windows(account.first, account.last, DRY_RUN_WINDOW_DAYS)) { + const c = await compareWindow(account.id, rank, w.from, w.to); + days += c.days; + changed += c.over1; + maxChange = Math.max(maxChange, c.maxDelta); + c.buckets.forEach((n, b) => (buckets[b] += n)); + } + daysCompared += days; + over1 += changed; + perAccount.push({ account: `#${i + 1}`, days, changedOver1Bpm: changed }); + } + return { + accounts: accounts.length, + daysCompared, + daysChangedOver1Bpm: over1, + maxChangeBpm: Math.round(maxChange * 10) / 10, + distribution: PULSE_DAY_MEAN_CHANGE_BUCKETS.map((bucket, b) => ({ + bucket, + days: buckets[b], + })), + perAccount, + }; +} + +// ─── Rebuild ────────────────────────────────────────────── + +/** + * Recompute every account's PULSE rollups inside the fold window, a year of + * one account at a time. Returns the accounts done and the rows written. + */ +export async function rebuildPulseRollups(): Promise<{ + accounts: number; + rows: number; +}> { + const floor = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + let rows = 0; + let accounts = 0; + for (const account of await pulseAccounts()) { + if (account.last < floor) continue; + const first = account.first < floor ? floor : account.first; + for (const w of windows(first, account.last, REBUILD_WINDOW_DAYS)) { + const done = await recomputeUserRollups(account.id, { + types: ["PULSE"], + from: w.from, + to: w.to, + statementTimeoutMs: STATEMENT_BUDGET_MS, + }); + rows += done.rowsUpserted; + } + accounts += 1; + } + return { accounts, rows }; +} + +// ─── Job ────────────────────────────────────────────────── + +async function runDryRun(): Promise { + const startedAt = new Date().toISOString(); + await writeDryRun({ + status: "running", + startedAt, + finishedAt: null, + result: null, + }); + try { + const result = await computePulseDayMeanDryRun(); + await writeDryRun({ + status: "done", + startedAt, + finishedAt: new Date().toISOString(), + result, + }); + return result; + } catch (err) { + await writeDryRun({ + status: "failed", + startedAt, + finishedAt: new Date().toISOString(), + result: null, + }).catch((writeErr) => + logCaught("pulse_day_mean.dry_run.status_write_failed", writeErr), + ); + throw err; + } +} + +export async function handlePulseDayMean( + jobs: Job[], +): Promise { + const action = jobs[0]?.data?.action === "rebuild" ? "rebuild" : "dry_run"; + return withBackgroundEvent("job.pulse_day_mean", async (evt) => { + const startedAt = Date.now(); + evt.addMeta("pulse_day_mean_action", action); + try { + const guarded = await withJobLock(PULSE_DAY_MEAN_QUEUE, async () => + action === "rebuild" + ? { kind: "rebuild" as const, ...(await rebuildPulseRollups()) } + : { kind: "dry_run" as const, result: await runDryRun() }, + ); + if (!guarded.ran) { + return jobDone({ outcome: "already_running" }); + } + const done = guarded.result; + if (done.kind === "rebuild") { + evt.addMeta("pulse_day_mean_accounts", done.accounts); + evt.addMeta("pulse_day_mean_rows", done.rows); + return jobDone({ + outcome: "rebuilt", + accounts_completed: done.accounts, + rows_upserted: done.rows, + duration_ms: Date.now() - startedAt, + }); + } + evt.addMeta("pulse_day_mean_accounts", done.result.accounts); + evt.addMeta("pulse_day_mean_days", done.result.daysCompared); + evt.addMeta("pulse_day_mean_changed", done.result.daysChangedOver1Bpm); + return jobDone({ + outcome: "compared", + accounts_completed: done.result.accounts, + processed: done.result.daysCompared, + duration_ms: Date.now() - startedAt, + }); + } catch (err) { + logCaught(`pulse_day_mean.${action}_failed`, err); + return jobFailed(`pulse day-mean ${action} failed`, err, { + duration_ms: Date.now() - startedAt, + }); + } + }); +} diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 319df60e9..3b94cd1a9 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -41,6 +41,7 @@ import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "@/lib/jobs/apple-health-import import { MEDICATION_INTAKE_IMPORT_SEND_OPTIONS } from "@/lib/jobs/medication-intake-import"; import { HEALTH_CONNECT_IMPORT_SEND_OPTIONS } from "@/lib/jobs/health-connect-import-worker"; import { MEASUREMENT_MAINTENANCE_SEND_OPTIONS } from "@/lib/jobs/measurement-maintenance"; +import { PULSE_DAY_MEAN_SEND_OPTIONS } from "@/lib/jobs/pulse-day-mean"; /** pg-boss's default job expiry, in seconds. */ export const DEFAULT_JOB_EXPIRE_SECONDS = 15 * 60; @@ -410,6 +411,23 @@ export const QUEUE_RUNTIME: Readonly> = { }, why: "Rebuilds every index of the largest table, one at a time.", }, + "pulse-day-mean": { + runtime: "long", + expireInSeconds: PULSE_DAY_MEAN_SEND_OPTIONS.expireInSeconds, + expiryVia: ["PULSE_DAY_MEAN_SEND_OPTIONS"], + stop: { + oneShot: + "A dry run or rebuild that stops part way is started again by the operator; the send sets retryLimit 0.", + }, + exclusive: { + marker: "retryLimit: 0", + at: { + file: "lib/jobs/pulse-day-mean.ts", + fn: "PULSE_DAY_MEAN_SEND_OPTIONS", + }, + }, + why: "Walks every account's pulse history a window at a time, twice over for a dry run.", + }, "medication-intake-import": { runtime: "long", expireInSeconds: MEDICATION_INTAKE_IMPORT_SEND_OPTIONS.expireInSeconds, diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 2e8d413b0..76e4ac419 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -83,6 +83,11 @@ import { handleMeasurementMaintenance, type MeasurementMaintenancePayload, } from "@/lib/jobs/measurement-maintenance"; +import { + PULSE_DAY_MEAN_QUEUE, + handlePulseDayMean, + type PulseDayMeanPayload, +} from "@/lib/jobs/pulse-day-mean"; import { HEALTH_CONNECT_IMPORT_QUEUE, HEALTH_CONNECT_IMPORT_CONCURRENCY, @@ -496,6 +501,8 @@ const allQueues = [ // cron: the admin route enqueues it. Without this entry the trigger // silently no-ops. MEASUREMENT_MAINTENANCE_QUEUE, + // v1.43 — the pulse day-mean dry run and rebuild, admin-triggered only. + PULSE_DAY_MEAN_QUEUE, // v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery enqueues // one job per user holding two live intake rows that snap to the same // canonical slot (the pre-fix REMINDER-pending + API-taken pair). Also @@ -835,6 +842,12 @@ const queuePolicies: QueuePolicyTable = { reason: "Fixed singleton key, admin-triggered. Two maintenance passes would rebuild the same indexes side by side; the route reports a suppressed send back as enqueued: false.", }, + // v1.43 — the pulse day-mean dry run and rebuild. + [PULSE_DAY_MEAN_QUEUE]: { + policy: "exclusive", + reason: + "Fixed singleton key, admin-triggered. A dry run beside a rebuild would compare against rollups being rewritten; the route answers 409 busy instead.", + }, // v1.42 — compaction-tombstone backlog purge. `short`, not `exclusive`: // a run that stops early sends its own follow-up while it is still active, // and `exclusive` would swallow that send. A boot's send while a run is @@ -1228,6 +1241,13 @@ export async function registerMaintenanceQueues( { localConcurrency: 1 }, handleMeasurementMaintenance, ); + // v1.43 — pulse day-mean dry run and rebuild, operator-triggered only. + await createAndWork( + boss, + PULSE_DAY_MEAN_QUEUE, + { localConcurrency: 1 }, + handlePulseDayMean, + ); // v1.32.1 (issue #588) — periodic orphan-ImportJob sweep. Single-flight: // the underlying `updateMany` is idempotent and two ticks racing the same // read-then-write pass would just be wasted work. diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index ed4dfa171..d20e764ce 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -123,6 +123,7 @@ const subsystemSurface = { "compaction-tombstone-purge": { audience: "system" }, "measurement-fold-repair": { audience: "system" }, "measurement-maintenance": { audience: "system" }, + "pulse-day-mean": { audience: "system" }, "medication-intake-import": { audience: "system" }, "intake-slot-dedup": { audience: "system" }, "mood-reminder-cleanup": { audience: "system" }, diff --git a/src/lib/measurements/__tests__/day-mean.test.ts b/src/lib/measurements/__tests__/day-mean.test.ts index f2da9e6fc..635aa76f6 100644 --- a/src/lib/measurements/__tests__/day-mean.test.ts +++ b/src/lib/measurements/__tests__/day-mean.test.ts @@ -1,17 +1,26 @@ -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it } from "vitest"; import { + combineHourCells, dayValue, dayValueRows, dayWeightedRows, foldMeanSql, - hourMeanRows, + hourWeightSql, isHourlyMeanTypeSql, readingsMean, + storedHourWeightSql, + weightedHourMeanSql, windowMeanSql, zoneDayFrame, SESSION_DAY_FRAME, } from "../day-mean"; +import { + activePulseDayMeanMethod, + hourWeight, + isPulseDayMeanMethod, + setActivePulseDayMeanMethod, +} from "../day-statistic"; const at = (iso: string) => new Date(iso); @@ -93,24 +102,109 @@ describe("readingsMean", () => { }); }); -describe("hourMeanRows / dayValueRows", () => { - it("collapse pulse to one row per hour / per day", () => { +describe("dayValueRows", () => { + it("collapses pulse to one row per day", () => { const rows = workoutDay("2026-03-02"); - const hours = hourMeanRows("PULSE", rows, "UTC"); - expect(hours.map((h) => h.value)).toEqual([150, 60, 60, 60]); - expect(hours[0].measuredAt.toISOString()).toBe("2026-03-02T10:00:00.000Z"); const days = dayValueRows("PULSE", rows, "UTC"); expect(days).toHaveLength(1); expect(days[0].value).toBeCloseTo(82.5, 10); + expect(days[0].measuredAt.toISOString()).toBe("2026-03-02T10:00:00.000Z"); }); - it("hand every other type back untouched", () => { + it("hands every other type back untouched", () => { const rows = workoutDay("2026-03-02"); - expect(hourMeanRows("BLOOD_GLUCOSE", rows, "UTC")).toBe(rows); expect(dayValueRows("BLOOD_GLUCOSE", rows, "UTC")).toBe(rows); }); }); +/** One sparse hour (a single reading at 120), then 23 dense hours at 60. */ +function sparseHourDay(day: string) { + const rows = [{ value: 120, measuredAt: at(`${day}T00:30:00Z`) }]; + for (let h = 1; h < 24; h++) { + for (let m = 0; m < 60; m += 10) { + rows.push({ + value: 60, + measuredAt: at( + `${day}T${String(h).padStart(2, "0")}:${String(m).padStart(2, "0")}:00Z`, + ), + }); + } + } + return rows; +} + +describe("pulse day-mean method", () => { + afterEach(() => setActivePulseDayMeanMethod("hour_mean_v1")); + + it("weighs a lone reading's hour a third under weighted_hour_mean_v2", () => { + const rows = sparseHourDay("2026-03-02"); + expect(dayValue("PULSE", rows, "UTC", "hour_mean_v1")).toBeCloseTo( + 62.5, + 10, + ); + const v2 = dayValue("PULSE", rows, "UTC", "weighted_hour_mean_v2"); + expect(Number(v2!.toFixed(1))).toBe(60.9); + expect(v2).toBeCloseTo((23 * 60 + 120 / 3) / (23 + 1 / 3), 10); + }); + + it("leaves a dense day where it was", () => { + // Every hour holds at least three readings, so every hour weighs one. + const dense = sparseHourDay("2026-03-02").slice(1); + expect(dayValue("PULSE", dense, "UTC", "weighted_hour_mean_v2")).toBe( + dayValue("PULSE", dense, "UTC", "hour_mean_v1"), + ); + }); + + it("follows the active method when the caller names none", () => { + const rows = sparseHourDay("2026-03-02"); + expect(activePulseDayMeanMethod()).toBe("hour_mean_v1"); + expect(dayValue("PULSE", rows, "UTC")).toBeCloseTo(62.5, 10); + setActivePulseDayMeanMethod("weighted_hour_mean_v2"); + expect(dayValue("PULSE", rows, "UTC")).toBeCloseTo( + (23 * 60 + 40) / (23 + 1 / 3), + 10, + ); + expect(readingsMean("PULSE", rows, "UTC")).toBeCloseTo( + (23 * 60 + 40) / (23 + 1 / 3), + 10, + ); + expect(dayValueRows("PULSE", rows, "UTC")[0].value).toBeCloseTo( + (23 * 60 + 40) / (23 + 1 / 3), + 10, + ); + }); + + 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].map((n) => hourWeight(n, "hour_mean_v1"))).toEqual([ + 1, 1, 1, + ]); + }); + + it("combines hour cells and skips empty ones", () => { + const cells = [ + { sum: 120, n: 1 }, + { sum: 360, n: 6 }, + { sum: 0, n: 0 }, + ]; + expect(combineHourCells(cells, "hour_mean_v1")).toBe(90); + expect(combineHourCells(cells, "weighted_hour_mean_v2")).toBeCloseTo( + (40 + 60) / (4 / 3), + 10, + ); + expect(combineHourCells([], "weighted_hour_mean_v2")).toBeNull(); + }); + + it("recognises only the two stored values", () => { + expect(isPulseDayMeanMethod("hour_mean_v1")).toBe(true); + expect(isPulseDayMeanMethod("weighted_hour_mean_v2")).toBe(true); + expect(isPulseDayMeanMethod("weighted")).toBe(false); + expect(isPulseDayMeanMethod(null)).toBe(false); + }); +}); + describe("SQL builders", () => { it("splice only the closed type list", () => { expect(isHourlyMeanTypeSql('m."type"')).toBe( @@ -142,4 +236,30 @@ describe("SQL builders", () => { }); expect(fold).toContain("ELSE SUM(c.total) / SUM(c.cnt) END"); }); + + it("build the hour weight per method and refuse a non-column", () => { + expect(hourWeightSql("h.n", "hour_mean_v1")).toBe("1.0"); + expect(hourWeightSql("h.n", "weighted_hour_mean_v2")).toBe( + "LEAST(1.0, (h.n)::double precision / 3.0)", + ); + expect(storedHourWeightSql("h.n")).toContain( + `"pulse_day_mean_method" FROM "app_settings"`, + ); + expect(weightedHourMeanSql("h.hour_mean", "1.0")).toBe( + "(SUM((h.hour_mean) * 1.0) / NULLIF(SUM(1.0), 0))", + ); + expect(() => hourWeightSql("n; DROP", "hour_mean_v1")).toThrow(); + expect(() => weightedHourMeanSql("x'", "1.0")).toThrow(); + }); + + it("weigh the per-row day weight by the method", () => { + const v1 = dayWeightedRows("cm", SESSION_DAY_FRAME, { + method: "hour_mean_v1", + }); + const v2 = dayWeightedRows("cm", SESSION_DAY_FRAME, { + method: "weighted_hour_mean_v2", + }); + expect(v1).not.toContain("LEAST"); + expect(v2).toContain("LEAST(1.0, (s.dm_hour_n)::double precision / 3.0)"); + }); }); diff --git a/src/lib/measurements/daily-series-read.ts b/src/lib/measurements/daily-series-read.ts index ef2b0659e..468921d95 100644 --- a/src/lib/measurements/daily-series-read.ts +++ b/src/lib/measurements/daily-series-read.ts @@ -16,6 +16,10 @@ import { type AggregateGrain, } from "@/lib/measurements/range-aggregation"; import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { + hourWeightSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; import { readTieredRollupSeries } from "@/lib/rollups/measurement-read-wmy"; import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; @@ -201,6 +205,11 @@ export async function readLiveBuckets(opts: { const rankRaw = Prisma.raw( buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'), ); + // The day of an hourly-mean type: its hours' means, each hour weighed by the + // active pulse day-mean method (`day-mean.ts`). Code-built SQL, no input. + const dayOfHours = Prisma.raw( + weightedHourMeanSql("h.hour_mean", hourWeightSql("h.n")), + ); const typeFilter = type ? Prisma.sql`AND m."type" = ${type}::measurement_type` : Prisma.empty; @@ -242,20 +251,21 @@ export async function readLiveBuckets(opts: { ${typeFilter} GROUP BY m."type", m."source", 3 ), - -- A day of a type with an activity-dependent sampling rate is the mean of - -- its local hours' means, each hour once, not the mean of its readings: a - -- workout hour holds hundreds of times as many readings as a resting one. + -- A day of a type with an activity-dependent sampling rate is the weighted + -- mean of its local hours' means, not the mean of its readings: a workout + -- hour holds hundreds of times as many readings as a resting one. -- Empty for every other type. hour_mean AS ( SELECT h."type", h."source", h.day, - AVG(h.hour_mean)::double precision AS day_mean + ${dayOfHours}::double precision AS day_mean FROM ( SELECT m."type", m."source", date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, - AVG(m."value") AS hour_mean + AVG(m."value") AS hour_mean, + COUNT(*) AS n FROM measurements m WHERE m."user_id" = ${userId} AND m."measured_at" >= ${from} diff --git a/src/lib/measurements/day-mean.ts b/src/lib/measurements/day-mean.ts index 9d90908bc..57ac46c89 100644 --- a/src/lib/measurements/day-mean.ts +++ b/src/lib/measurements/day-mean.ts @@ -2,10 +2,14 @@ * The one place that turns readings into a day value and a window mean for the * types in `HOURLY_MEAN_DAY_TYPES` (see `day-statistic.ts`). * - * For such a type a day is the mean of its hours' means, each hour weighing - * one, and a window of days is the mean of its day values, each day weighing - * one. Count, min and max stay over every reading. Every other type keeps the - * plain mean over its readings, computed exactly as before. + * For such a type a day is the weighted mean of its hours' means and a + * window of days the mean of its day values, each day weighing one. How an + * hour weighs is the pulse day-mean method (`day-statistic.ts`): one under + * `hour_mean_v1`, `min(1, n / 3)` for an hour of `n` readings under + * `weighted_hour_mean_v2`. Every helper applies this process's active method + * unless the caller names one. Count, min and max stay over every reading. + * Every other type keeps the plain mean over its readings, computed exactly as + * before. * * "Day" and "hour" are the reader's own frame: a reader that buckets by the * user's local day passes that zone, a reader that buckets by the session's @@ -15,14 +19,17 @@ * Two shapes are offered, one per kind of reader: * * - TypeScript, for readers that hold the readings in memory: - * `readingsMean`, `dayValue` and `dayValueRows`. + * `readingsMean`, `dayValue` and `dayValueRows`, and `combineHourCells` + * for a reader that already holds per-hour sums. * - SQL, for readers that aggregate in Postgres: `dayWeightedRows` attaches - * a `day_weight` to every reading (1 / (readings in its hour × hours in its - * day) for an hourly-mean type, 1 otherwise), so a day's weights add up to - * one and `SUM(value * day_weight) / SUM(day_weight)` is the mean of the - * day values over any window. `windowMeanSql` and `foldMeanSql` build the - * mean expressions, falling back to the unchanged plain mean for every - * other type. + * a `day_weight` to every reading (its hour's weight shared among the + * hour's readings, over the day's total hour weight, for an hourly-mean + * type; 1 otherwise), so a day's weights add up to one and + * `SUM(value * day_weight) / SUM(day_weight)` is the mean of the day + * values over any window. `windowMeanSql` and `foldMeanSql` build the mean + * expressions, falling back to the unchanged plain mean for every other + * type. A reader that groups by hour itself weighs its hours with + * `hourWeightSql` (or `storedHourWeightSql`) and `weightedHourMeanSql`. * * Type lists are closed code constants checked against an enum-shaped regex * before they are spliced; every value (user id, zone, bounds) stays a bound @@ -31,7 +38,14 @@ import { Prisma } from "@/generated/prisma/client"; import { tzOffsetMinutes } from "@/lib/tz/format"; -import { HOURLY_MEAN_DAY_TYPES, usesHourlyMeanDay } from "./day-statistic"; +import { + activePulseDayMeanMethod, + HOURLY_MEAN_DAY_TYPES, + hourWeight, + type PulseDayMeanMethod, + SPARSE_HOUR_FULL_WEIGHT, + usesHourlyMeanDay, +} from "./day-statistic"; const MS_PER_MINUTE = 60_000; const MS_PER_HOUR = 3_600_000; @@ -66,10 +80,32 @@ function plainMean(rows: readonly DayReading[]): number | null { return rows.reduce((s, r) => s + r.value, 0) / rows.length; } -/** Mean of the per-hour means of readings that share one day. */ +/** + * A day value from its hours' reading sums and counts: the mean of the hours' + * means, each hour weighed by `hourWeight` under `method`. `null` when no hour + * holds a reading. The one TypeScript place the weighting lives; a reader + * that already holds per-hour sums (`daily-stats.ts`) calls it directly. + */ +export function combineHourCells( + hours: Iterable<{ sum: number; n: number }>, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), +): number | null { + let weighted = 0; + let weights = 0; + for (const h of hours) { + if (h.n <= 0) continue; + const w = hourWeight(h.n, method); + weighted += w * (h.sum / h.n); + weights += w; + } + return weights > 0 ? weighted / weights : null; +} + +/** The weighted mean of the per-hour means of readings that share one day. */ function meanOfHourMeans( rows: readonly DayReading[], localMs: (at: Date) => number, + method: PulseDayMeanMethod, ): number { const hours = new Map(); for (const r of rows) { @@ -82,9 +118,7 @@ function meanOfHourMeans( hours.set(hour, { sum: r.value, n: 1 }); } } - let sum = 0; - for (const h of hours.values()) sum += h.sum / h.n; - return sum / hours.size; + return combineHourCells(hours.values(), method) as number; } /** Group readings by local day (days since the epoch in `tz`). */ @@ -110,10 +144,11 @@ export function dayValue( type: string, rows: readonly DayReading[], tz: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), ): number | null { if (rows.length === 0) return null; if (!usesHourlyMeanDay(type)) return plainMean(rows); - return meanOfHourMeans(rows, makeLocalMs(tz)); + return meanOfHourMeans(rows, makeLocalMs(tz), method); } /** @@ -126,6 +161,7 @@ export function readingsMean( type: string, rows: readonly DayReading[], tz: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), ): number | null { if (rows.length === 0) return null; if (!usesHourlyMeanDay(type)) return plainMean(rows); @@ -133,43 +169,12 @@ export function readingsMean( let sum = 0; let days = 0; for (const dayRows of groupByLocalDay(rows, localMs).values()) { - sum += meanOfHourMeans(dayRows, localMs); + sum += meanOfHourMeans(dayRows, localMs, method); days += 1; } return sum / days; } -/** - * Readings of an hourly-mean type replaced by one row per local hour carrying - * the hour's mean, stamped at the hour's first reading. A reader that already - * folds its rows into day buckets by plain mean (and a window by mean of - * those) then yields the hourly-mean day without knowing the statistic. Every - * other type comes back untouched (the same array). - */ -export function hourMeanRows( - type: string, - rows: readonly T[], - tz: string, -): readonly DayReading[] { - if (!usesHourlyMeanDay(type) || rows.length === 0) return rows; - const localMs = makeLocalMs(tz); - const hours = new Map(); - for (const r of rows) { - const hour = Math.floor(localMs(r.measuredAt) / MS_PER_HOUR); - const acc = hours.get(hour); - if (acc) { - acc.sum += r.value; - acc.n += 1; - if (r.measuredAt < acc.at) acc.at = r.measuredAt; - } else { - hours.set(hour, { sum: r.value, n: 1, at: r.measuredAt }); - } - } - return [...hours.values()] - .map((h) => ({ value: h.sum / h.n, measuredAt: h.at })) - .sort((a, b) => a.measuredAt.getTime() - b.measuredAt.getTime()); -} - /** * Readings of an hourly-mean type replaced by one row per local day carrying * the day value, stamped at the day's first reading; a reader that folds rows @@ -180,6 +185,7 @@ export function dayValueRows( type: string, rows: readonly T[], tz: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), ): readonly DayReading[] { if (!usesHourlyMeanDay(type) || rows.length === 0) return rows; const localMs = makeLocalMs(tz); @@ -187,7 +193,10 @@ export function dayValueRows( for (const dayRows of groupByLocalDay(rows, localMs).values()) { let first = dayRows[0].measuredAt; for (const r of dayRows) if (r.measuredAt < first) first = r.measuredAt; - out.push({ value: meanOfHourMeans(dayRows, localMs), measuredAt: first }); + out.push({ + value: meanOfHourMeans(dayRows, localMs, method), + measuredAt: first, + }); } return out.sort((a, b) => a.measuredAt.getTime() - b.measuredAt.getTime()); } @@ -237,10 +246,55 @@ function frameTs(frame: DayFrame, column: string): string { : `((${column} AT TIME ZONE 'UTC') AT TIME ZONE ${frame.tzSql})`; } +/** A column or expression of the caller's own query, never input. */ +const SQL_EXPR_RE = /^[A-Za-z0-9_."() ]+$/; + +/** + * SQL weight of an hour holding `n` readings under `method` (see + * `hourWeight`): `1.0` for `hour_mean_v1`, `LEAST(1, n / 3)` for + * `weighted_hour_mean_v2`. `n` is a column of the caller's query. + */ +export function hourWeightSql( + n: string, + method: PulseDayMeanMethod = activePulseDayMeanMethod(), +): string { + if (!SQL_EXPR_RE.test(n)) { + throw new Error(`hour count must be a column expression: ${n}`); + } + return method === "weighted_hour_mean_v2" + ? `LEAST(1.0, (${n})::double precision / ${SPARSE_HOUR_FULL_WEIGHT}.0)` + : "1.0"; +} + +/** + * `hourWeightSql` under the method stored in `app_settings`, read inside the + * statement itself. For a writer whose output persists (the rollup tier): it + * must follow the setting the moment the apply step changes it, not this + * process's mirror of it, which trails by up to a minute. + */ +export function storedHourWeightSql(n: string): string { + return `(CASE WHEN (SELECT st."pulse_day_mean_method" FROM "app_settings" st WHERE st."id" = 'singleton') = 'weighted_hour_mean_v2' + THEN ${hourWeightSql(n, "weighted_hour_mean_v2")} + ELSE 1.0 END)`; +} + +/** + * Aggregate over per-hour rows: the day value as the weighted mean of the + * hours' means, `SUM(hourMean * weight) / SUM(weight)`. `weight` comes from + * `hourWeightSql` or `storedHourWeightSql`. + */ +export function weightedHourMeanSql(hourMean: string, weight: string): string { + if (!SQL_EXPR_RE.test(hourMean)) { + throw new Error(`hour mean must be a column expression: ${hourMean}`); + } + return `(SUM((${hourMean}) * ${weight}) / NULLIF(SUM(${weight}), 0))`; +} + function weightedRowsSql( source: string, frame: DayFrame, bySource: boolean, + method: PulseDayMeanMethod, ): string { if (!/^[a-z_][a-z0-9_]*$/.test(source)) { throw new Error(`source must be a CTE name: ${source}`); @@ -251,8 +305,11 @@ function weightedRowsSql( // join as a nested loop and a year of pulse ran into the statement timeout. // // `dm_hour_n` counts the readings in a row's hour and `dm_hour_first` marks - // one row per hour, so the sum of `dm_hour_first` over a day is the day's - // hour count. Both columns ride along on the row; every caller aggregates. + // one row per hour, so the sum of `dm_hour_first * w` over a day is the + // day's total hour weight (its hour count under `hour_mean_v1`). A row's + // weight is its hour's weight shared among the hour's readings, divided by + // that total, so a day's weights still add up to one. Both columns ride + // along on the row; every caller aggregates. const partition = (alias: string, unit: "hour" | "day") => { const keys = bySource ? `${alias}."type", ${alias}."source"` @@ -263,7 +320,7 @@ function weightedRowsSql( return `( SELECT s.*, (CASE WHEN ${isHourlyMeanTypeSql('s."type"')} - THEN 1.0 / (s.dm_hour_n * SUM(s.dm_hour_first) OVER (${partition("s", "day")})) + THEN ${hourWeightSql("s.dm_hour_n", method)} / (s.dm_hour_n * SUM(s.dm_hour_first * ${hourWeightSql("s.dm_hour_n", method)}) OVER (${partition("s", "day")})) ELSE 1.0 END)::double precision AS day_weight FROM ( SELECT t.*, @@ -286,9 +343,14 @@ function weightedRowsSql( export function dayWeightedRows( source: string, frame: DayFrame, - opts: { bySource?: boolean } = {}, + opts: { bySource?: boolean; method?: PulseDayMeanMethod } = {}, ): string { - return weightedRowsSql(source, frame, opts.bySource === true); + return weightedRowsSql( + source, + frame, + opts.bySource === true, + opts.method ?? activePulseDayMeanMethod(), + ); } /** @@ -298,11 +360,17 @@ export function dayWeightedRows( export function dayWeightedRowsSql( source: string, tz: string | null, - opts: { bySource?: boolean } = {}, + opts: { bySource?: boolean; method?: PulseDayMeanMethod } = {}, ): Prisma.Sql { + const method = opts.method ?? activePulseDayMeanMethod(); if (tz === null) { return Prisma.raw( - weightedRowsSql(source, SESSION_DAY_FRAME, opts.bySource === true), + weightedRowsSql( + source, + SESSION_DAY_FRAME, + opts.bySource === true, + method, + ), ); } // Split around a placeholder so the zone binds as a parameter. @@ -311,6 +379,7 @@ export function dayWeightedRowsSql( source, { kind: "zone", tzSql: marker }, opts.bySource === true, + method, ).split(marker); let sql = Prisma.raw(pieces[0]); for (const piece of pieces.slice(1)) { diff --git a/src/lib/measurements/day-statistic.ts b/src/lib/measurements/day-statistic.ts index 2b42cf4f0..2baa12799 100644 --- a/src/lib/measurements/day-statistic.ts +++ b/src/lib/measurements/day-statistic.ts @@ -30,3 +30,73 @@ export type WindowWeighting = "count" | "day"; export function windowWeighting(type: string): WindowWeighting { return usesHourlyMeanDay(type) ? "day" : "count"; } + +// ─── Pulse day-mean method (#1107 follow-up) ────────────── + +/** + * How the hours of a pulse day weigh against each other. + * + * - `hour_mean_v1`: every hour that holds a reading weighs one, the + * statistic up to v1.42. One stray reading in an otherwise empty hour + * then counts as much as an hour of steady sampling. + * - `weighted_hour_mean_v2`: an hour with `n` readings weighs + * `min(1, n / SPARSE_HOUR_FULL_WEIGHT)`, so an hour reaches full weight + * at three readings and a lone reading weighs a third. A dense hour still + * weighs one, so the workout bias the hourly mean removed stays removed. + * + * The operator switches it once, after a dry run, from the admin console; + * `app_settings.pulse_day_mean_method` holds it. + */ +export const PULSE_DAY_MEAN_METHODS = [ + "hour_mean_v1", + "weighted_hour_mean_v2", +] as const; +export type PulseDayMeanMethod = (typeof PULSE_DAY_MEAN_METHODS)[number]; + +export const DEFAULT_PULSE_DAY_MEAN_METHOD: PulseDayMeanMethod = "hour_mean_v1"; + +/** Readings at which an hour reaches full weight under `weighted_hour_mean_v2`. */ +export const SPARSE_HOUR_FULL_WEIGHT = 3; + +export function isPulseDayMeanMethod( + value: unknown, +): value is PulseDayMeanMethod { + return ( + typeof value === "string" && + (PULSE_DAY_MEAN_METHODS as readonly string[]).includes(value) + ); +} + +/** The weight of an hour holding `n` readings under `method`. */ +export function hourWeight(n: number, method: PulseDayMeanMethod): number { + if (method === "weighted_hour_mean_v2") { + return Math.min(1, n / SPARSE_HOUR_FULL_WEIGHT); + } + return 1; +} + +/** + * The method this process applies. Every reader computes synchronously, so + * the stored setting is mirrored here: loaded at boot and refreshed every + * minute (`pulse-day-mean-method.ts`), set at once by the apply step in the + * process that ran it. Pinned on `globalThis` because the boot hook and the + * route handlers are separate bundles of one process. The default until the + * first load is the v1.42 statistic, which is also what the column defaults + * to. + */ +const METHOD_SYMBOL = Symbol.for("healthlog.pulseDayMeanMethod"); + +type GlobalWithMethod = typeof globalThis & { + [METHOD_SYMBOL]?: PulseDayMeanMethod; +}; + +export function activePulseDayMeanMethod(): PulseDayMeanMethod { + return ( + (globalThis as GlobalWithMethod)[METHOD_SYMBOL] ?? + DEFAULT_PULSE_DAY_MEAN_METHOD + ); +} + +export function setActivePulseDayMeanMethod(method: PulseDayMeanMethod): void { + (globalThis as GlobalWithMethod)[METHOD_SYMBOL] = method; +} diff --git a/src/lib/measurements/pulse-day-mean-method.ts b/src/lib/measurements/pulse-day-mean-method.ts new file mode 100644 index 000000000..91be70cbc --- /dev/null +++ b/src/lib/measurements/pulse-day-mean-method.ts @@ -0,0 +1,70 @@ +/** + * Keeps this process's mirror of the pulse day-mean method in step with + * `app_settings.pulse_day_mean_method` (#1107 follow-up). + * + * Every pulse reader computes synchronously through `day-mean.ts`, so the + * method cannot be read per call; it is mirrored instead + * (`activePulseDayMeanMethod`). `instrumentation.ts` loads it once at boot and + * refreshes it every minute in every process, web and worker alike. The apply + * step sets it at once in the process that ran it; any other process follows + * within a minute. The rollup tier does not use the mirror: it reads the + * setting inside its own statement (`storedHourWeightSql`). + * + * A failed read keeps the mirror as it is: the setting changes once, by hand, + * and a database blip must not flip a process back to the old statistic. + */ +import type { Prisma } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; + +import { + DEFAULT_PULSE_DAY_MEAN_METHOD, + isPulseDayMeanMethod, + setActivePulseDayMeanMethod, + type PulseDayMeanMethod, +} from "./day-statistic"; + +/** How often a process re-reads the stored method. */ +const REFRESH_MS = 60_000; + +const TIMER_SYMBOL = Symbol.for("healthlog.pulseDayMeanMethodTimer"); + +type GlobalWithTimer = typeof globalThis & { + [TIMER_SYMBOL]?: ReturnType; +}; + +type SettingsReader = { + appSettings: Pick; +}; + +/** The stored method; the column default when there is no settings row. */ +export async function loadPulseDayMeanMethod( + db: SettingsReader = prisma, +): Promise { + const row = await db.appSettings.findUnique({ + where: { id: "singleton" }, + select: { pulseDayMeanMethod: true }, + }); + const stored = row?.pulseDayMeanMethod; + return isPulseDayMeanMethod(stored) ? stored : DEFAULT_PULSE_DAY_MEAN_METHOD; +} + +async function refresh(): Promise { + try { + setActivePulseDayMeanMethod(await loadPulseDayMeanMethod()); + } catch { + // Keep the mirror; the next tick tries again. + } +} + +/** + * Load the stored method now and keep re-reading it. Idempotent per process; + * the timer does not hold the process open. + */ +export async function startPulseDayMeanMethodSync(): Promise { + await refresh(); + const g = globalThis as GlobalWithTimer; + if (g[TIMER_SYMBOL]) return; + const timer = setInterval(() => void refresh(), REFRESH_MS); + timer.unref?.(); + g[TIMER_SYMBOL] = timer; +} diff --git a/src/lib/openapi/error-codes.ts b/src/lib/openapi/error-codes.ts index 8560c5a3f..83045e37e 100644 --- a/src/lib/openapi/error-codes.ts +++ b/src/lib/openapi/error-codes.ts @@ -256,6 +256,10 @@ export const ERROR_CODE_CATALOGUE: Readonly> = "managed_profile.handover.unknown_guardian", "managed_profile.not_found", ], + maintenance: [ + "maintenance.pulseDayMean.busy", + "maintenance.pulseDayMean.dryRunRequired", + ], measurement: [ "measurement.batch.invalid", "measurement.batch.source_not_permitted", diff --git a/src/lib/query-keys/admin.ts b/src/lib/query-keys/admin.ts index f3eeda9b7..dbc98647d 100644 --- a/src/lib/query-keys/admin.ts +++ b/src/lib/query-keys/admin.ts @@ -53,6 +53,8 @@ export const adminKeys = { adminEncryptionStatus: () => ["admin", "encryption", "status"] as const, adminMeasurementMaintenance: () => ["admin", "maintenance", "measurements"] as const, + /** v1.43 — the pulse day-mean method, its dry run and apply step. */ + adminPulseDayMean: () => ["admin", "maintenance", "pulse-day-mean"] as const, /** v1.40.0 — the "Back up your encryption key" step (admin + dashboard banner). */ adminEncryptionKeyBackup: () => ["admin", "encryption", "key-backup"] as const, diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 8f6643007..18b335740 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -43,6 +43,10 @@ import { utcBucketEnd, utcBucketStart, } from "@/lib/rollups/measurement-read"; +import { + storedHourWeightSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import type { @@ -588,16 +592,17 @@ async function runRollupAggregate( /** * For a type whose DAY value is the mean of its per-hour means (see - * `day-statistic.ts`), replace the bucket's `mean`: a DAY bucket's is the mean - * of its hours' means, a WEEK / MONTH / YEAR bucket's the mean of its days' - * values, each day once. The coarse rows are not read for their mean (every + * `day-statistic.ts`), replace the bucket's `mean`: a DAY bucket's is the + * weighted mean of its hours' means under the stored pulse day-mean method, + * a WEEK / MONTH / YEAR bucket's the mean of its days' values, each day once. The coarse rows are not read for their mean (every * coarse read folds the DAY tier, `readCanonicalRollupBuckets`), but a stored * row should not contradict the tier it is folded from. Only `mean` changes: * `count`, `min`, `max` and the sums the regression composes from stay over * every reading, so the band and the windowed slope are untouched. * * Hours and days are UTC, matching the UTC day this tier buckets by; migration - * 0372 applies the same statistic to rows stored before it. + * 0372 applies the same statistic to rows stored before it, and the admin + * apply step of the day-mean method rebuilds the PULSE rows it changes. */ async function applyHourlyMeanDay( db: Pick, @@ -627,6 +632,12 @@ async function applyHourlyMeanDay( } } const typeList = types.map((t) => `'${t}'::"measurement_type"`).join(","); + // The stored method, read inside the statement: a rollup outlives this + // process's mirror of the setting, so it follows the setting itself. + const dayOfHours = weightedHourMeanSql( + "h.hour_mean", + storedHourWeightSql("h.n"), + ); const means = await db.$queryRawUnsafe< Array<{ type: string; source: string; bucket_start: Date; mean: number }> >( @@ -638,11 +649,12 @@ async function applyHourlyMeanDay( FROM ( SELECT h."type", h."source", date_trunc('day', h.hr) AS day, - AVG(h.hour_mean) AS day_mean + ${dayOfHours} AS day_mean FROM ( SELECT m."type", m."source", date_trunc('hour', m."measured_at") AS hr, - AVG(m."value") AS hour_mean + AVG(m."value") AS hour_mean, + COUNT(*) AS n FROM measurements m WHERE m."user_id" = $1 AND m."type" IN (${typeList}) diff --git a/src/lib/validations/pulse-day-mean.ts b/src/lib/validations/pulse-day-mean.ts index 39a837534..f83a65269 100644 --- a/src/lib/validations/pulse-day-mean.ts +++ b/src/lib/validations/pulse-day-mean.ts @@ -1,14 +1,82 @@ /** * v1.43 (#1107 follow-up) — the admin pulse day-mean dry run and apply step. + * + * The status shape is the admin console's own and stays out of the published + * API description: the admin routes are not part of the public contract. */ import { z } from "zod/v4"; -export const PULSE_DAY_MEAN_METHODS = [ - "hour_mean_v1", - "weighted_hour_mean_v2", -] as const; +import type { PulseDayMeanMethod } from "@/lib/measurements/day-statistic"; export const pulseDayMeanActionSchema = z.strictObject({ action: z.enum(["dry_run", "apply"]), }); export type PulseDayMeanActionInput = z.infer; + +/** 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; + +/** + * What a dry run found. Accounts are numbered in the order they were created + * (`"#1"`, `"#2"`, …), never named or identified. + */ +const dryRunResultSchema = z.object({ + accounts: z.number().int().nonnegative(), + daysCompared: z.number().int().nonnegative(), + daysChangedOver1Bpm: z.number().int().nonnegative(), + maxChangeBpm: z.number().nonnegative(), + distribution: z.array( + z.object({ + bucket: z.enum(PULSE_DAY_MEAN_CHANGE_BUCKETS), + days: z.number().int().nonnegative(), + }), + ), + perAccount: z.array( + z.object({ + account: z.string(), + days: z.number().int().nonnegative(), + changedOver1Bpm: z.number().int().nonnegative(), + }), + ), +}); +export type PulseDayMeanDryRunResult = z.infer; + +/** `app_settings.pulse_day_mean_dry_run_json`: the last dry run. */ +export const pulseDayMeanDryRunSchema = z.object({ + status: z.enum(["queued", "running", "done", "failed"]), + startedAt: z.string(), + finishedAt: z.string().nullable(), + result: dryRunResultSchema.nullable(), +}); +export type PulseDayMeanDryRun = z.infer; + +/** `GET /api/admin/maintenance/pulse-day-mean`. */ +export interface PulseDayMeanStatus { + method: PulseDayMeanMethod; + appliedAt: string | null; + lastDryRun: PulseDayMeanDryRun | null; +} + +/** How old a finished dry run may be for the apply step to accept it. */ +const DRY_RUN_FRESH_MS = 7 * 24 * 60 * 60 * 1000; + +/** + * A finished dry run from the last seven days: the apply step's condition, + * enforced by the route and mirrored by the admin card's button. + */ +export function hasFreshDryRun( + status: Pick, + now: Date = new Date(), +): boolean { + const run = status.lastDryRun; + if (!run || run.status !== "done" || run.finishedAt === null) return false; + const finished = Date.parse(run.finishedAt); + return ( + Number.isFinite(finished) && now.getTime() - finished <= DRY_RUN_FRESH_MS + ); +} diff --git a/tests/integration/pulse-day-mean-method.test.ts b/tests/integration/pulse-day-mean-method.test.ts new file mode 100644 index 000000000..9aa1dff77 --- /dev/null +++ b/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[]; + +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> { + const tables = await prisma.$queryRaw>` + 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 = {}; + for (const { name } of tables) { + if (!/^[a-z0-9_]+$/.test(name)) throw new Error(`table name ${name}`); + const [r] = await prisma.$queryRawUnsafe>( + `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 }>>` + 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) { + 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); +});