From b20f1b5c6616eb69d60d12c789936237f1fc3710 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 21 Jul 2026 13:08:28 +0200 Subject: [PATCH] fix: harden imports and daily medication state (#577) --- CHANGELOG.md | 18 ++ docs/api/openapi.yaml | 35 ++- docs/integrations/apple-health.md | 77 +++-- messages/de.json | 1 + messages/en.json | 1 + messages/es.json | 1 + messages/fr.json | 1 + messages/it.json | 1 + messages/pl.json | 1 + package.json | 2 +- .../migration.sql | 27 ++ prisma/schema.prisma | 23 ++ public/sw.js | 2 +- src/__tests__/i18n-reverse-coverage.test.ts | 2 +- .../__tests__/route.test.ts | 27 +- .../admin/import-apple-health-export/route.ts | 7 +- .../__tests__/route.test.ts | 24 +- .../api/import/apple-health-export/route.ts | 9 +- src/app/api/measurements/batch/route.ts | 3 + .../settings/__tests__/import-panel.test.tsx | 19 +- .../import-panel/apple-health-import-card.tsx | 26 ++ src/lib/ai/__tests__/openai-client.test.ts | 115 +++++++ src/lib/ai/openai-capabilities.ts | 67 +++++ src/lib/ai/openai-client.ts | 20 +- src/lib/daily/__tests__/digest.test.ts | 105 ++++++- src/lib/daily/digest.ts | 135 +++++---- .../dashboard/__tests__/meds-today.test.ts | 58 ++++ src/lib/dashboard/meds-today.ts | 52 ++-- .../apple-health-import-legacy-bridge.test.ts | 92 ++++++ .../apple-health-import-send-policy.test.ts | 73 ++++- src/lib/jobs/apple-health-import-worker.ts | 74 ++++- src/lib/jobs/reminder/register-maintenance.ts | 28 +- .../drain-per-sample-cumulative.test.ts | 49 ++- .../import-apple-health-export.test.ts | 282 ++++++++++++++++-- .../reconcile-external-measurement.test.ts | 164 ++++++++++ .../drain-per-sample-cumulative.ts | 117 ++++---- .../import-apple-health-export.ts | 167 ++++++++--- .../reconcile-external-measurement.ts | 52 +++- .../scheduling/__tests__/next-due.test.ts | 43 +++ src/lib/medications/scheduling/next-due.ts | 73 ++++- src/lib/openapi/routes/insights/schemas.ts | 13 +- .../apple-health-import-reconcile.test.ts | 67 ++++- .../drain-per-sample-cumulative.test.ts | 91 ++++++ .../measurement-identity-reconcile.test.ts | 124 ++++++++ tests/integration/measurements-batch.test.ts | 181 +++++++++++ 45 files changed, 2270 insertions(+), 279 deletions(-) create mode 100644 prisma/migrations/0263_apple_health_aggregate_authority/migration.sql create mode 100644 src/lib/ai/openai-capabilities.ts create mode 100644 src/lib/jobs/__tests__/apple-health-import-legacy-bridge.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 6051b83f8..de77c8a26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,24 @@ ## [Unreleased] +## [1.31.7] — 2026-07-20 + +- **Current OpenAI reasoning models use their supported request contract.** + Official GPT-5, o1, o3, and o4 models receive `max_completion_tokens` + without unsupported `temperature` or `seed` fields; compatible custom + gateways retain the legacy parameters they advertise. +- **Apple Health XML cumulative totals no longer add overlapping sources.** + Imports aggregate each source/day independently and select the highest + source subtotal, while explicit provenance prevents XML estimates or late + legacy samples from overwriting authoritative native HealthKit statistics. + Existing legacy aggregates can be repaired by re-uploading their archive. +- **Every currently actionable medication reaches Today.** Equal-time doses no + longer collapse to one row, overdue doses win deterministic ties, and weekly + or custom intake windows use the scheduling engine's real availability + boundary. The shared digest fixes both the web start page and iOS Home. + +Migration `0263_apple_health_aggregate_authority`. No breaking changes. + ## [1.31.6] — 2026-07-21 - **The hardened runtime image can run database migrations again.** The diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index f2bdbbb89..41a360d09 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.31.6 + version: 1.31.7 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. @@ -25024,6 +25024,28 @@ components: anyOf: - type: string - type: "null" + dueCandidates: + type: array + items: + type: object + properties: + medicationId: + type: string + medicationName: + type: string + dueAt: + type: string + overdue: + type: boolean + availableFrom: + type: string + required: + - medicationId + - medicationName + - dueAt + - overdue + - availableFrom + additionalProperties: false required: - activeCount - scheduledToday @@ -25034,12 +25056,11 @@ components: - nextDueMedicationName - nextDueMedicationId additionalProperties: false - description: "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), and - the earliest next-due slot across active medications. `nextDueOverdue: true` marks an OPEN overdue slot - (anchor passed, still inside its catch-up band). `nextDueMedicationId` deep-links a consumer straight to the - overdue medication's card. Staleness contract: the snapshot is cache-served, so a `nextDueAt` in the past - with `nextDueOverdue: false` means the anchor passed after the snapshot was built — render the plain day - summary, never an overdue state." + description: "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), every + active medication's current display-due candidate, and legacy scalar fields projected from candidate zero. + `overdue: true` marks an OPEN overdue slot (anchor passed, still inside its catch-up band); `availableFrom` + is the canonical cadence- and dose-window-derived attribution-band start. `dueCandidates` is optional so + older cached snapshots remain valid. Medication ids let consumers deep-link straight to the relevant card." healthScore: anyOf: - type: object diff --git a/docs/integrations/apple-health.md b/docs/integrations/apple-health.md index 4d80057fc..0b3c4209d 100644 --- a/docs/integrations/apple-health.md +++ b/docs/integrations/apple-health.md @@ -112,29 +112,50 @@ mapped set (workout routes, ECG voltage traces) is counted in ### Cumulative HK type → daily aggregate Cumulative quantity types — steps, active energy, walking/running -distance, flights climbed — get **collapsed into one row per -user-local day** to match the iOS daily-aggregation convention. +distance, flights climbed, time in daylight, and fall count — get +**collapsed into one estimated row per user-local day**. `export.xml` +contains source-specific records, not HealthKit's canonical daily +statistics, so the importer does not sum different devices or apps +into a source-blind total. +- Records are grouped by type, local day, and a SHA-256 hash of the + available source name, source version, and device attributes. + Values sum within each identified source-day; the largest + source-day subtotal is selected deterministically. If every source + attribute is absent, records remain separate rather than being + trusted as one source. Raw source and device labels are not + persisted. +- The row records `EXPORT_XML_SOURCE_MAX` provenance, its distinct + contributor count, and the selected non-identifying source hash. - External ID format: `stats::` (e.g. `stats:HKQuantityTypeIdentifierStepCount:2026-05-15`). - The day boundary respects the user's timezone preference. The worker defaults to `Europe/Berlin` when no preference is set - (`src/lib/jobs/apple-health-import-worker.ts:132-133`). -- A second import of the same archive re-folds the same days onto - the same external IDs — the values update in place rather than - duplicating. + (`src/lib/jobs/apple-health-import-worker.ts`). +- XML intervals are not reconstructed or split. A record crossing + midnight is assigned using its mapped timestamp and remains covered + by the explicit estimate contract. +- A second import re-folds the same days onto the same external IDs. + Equal-authority estimates update idempotently rather than duplicate. + +The terminal job result exposes +`cumulativeEstimates: { days, rows }`; the web import card warns when +one or more cumulative days used this estimate path. ### Re-import idempotency The two-axis idempotency story: 1. **File-level** — re-uploading the same `export.zip` (same SHA-256 - of bytes) short-circuits to the previous job before the parser - even starts. + of bytes) short-circuits to the previous job only when that job + used the current parser revision. Existing jobs are revision 1; + new jobs use revision 2, so a pre-fix successful archive can be + deliberately processed again under the corrected aggregation. 2. **Record-level** — when an older `export.zip` is re-exported from iOS (same history, slightly newer device timestamps), every - record's external ID stays stable. The upsert path matches on - `(userId, externalId)` and updates instead of inserting. + record's external ID stays stable. The serialized reconciliation + path matches the canonical identity and updates instead of + inserting. Practically: a user who exports monthly and re-imports each archive sees fresh data merged in without duplicates. The job result line @@ -150,14 +171,34 @@ is identical to the user-facing flow. The `ImportJob` row carries `triggeredByAdminId = admin.id` so the status endpoint admits both the target user and the triggering -admin. Idempotency is scoped by target user — an admin re-uploading -the same archive for the same user resolves to the previous job. +admin. Idempotency is scoped by target user and parser revision — an +admin re-uploading the same archive for the same user resolves to the +previous current-revision job. Typical use: migrating a friend's history into their HealthLog account when they cannot run the import themselves, or backfilling a household member's data after the admin received the `export.zip` out of band. +## Aggregate authority and recovery + +Native HealthKit daily statistics submitted by the iOS batch endpoint +are canonical and carry `HEALTHKIT_STATISTICS` provenance. XML +source-maximum estimates carry `EXPORT_XML_SOURCE_MAX`. Both use the +same serialized reconciler with this ordering: + +`LEGACY_UNKNOWN < EXPORT_XML_SOURCE_MAX < HEALTHKIT_STATISTICS` + +A later XML import cannot overwrite native statistics. Native +statistics repair an XML estimate or legacy row in either arrival +order, and every material reconciled update or resurrection increments +`syncVersion`. Existing Apple Health `stats:*` rows are migrated to +`LEGACY_UNKNOWN`; they are never guessed or heuristically promoted. +To recover an already-corrupted legacy total, the user must re-upload +the original archive under parser revision 2 or sync native HealthKit +statistics. Without either input, the correct per-source total cannot +be reconstructed from the stored aggregate. + ## Source-priority interaction HealthLog's source ladder is **APPLE_HEALTH ≻ WITHINGS ≻ MANUAL ≻ @@ -169,12 +210,12 @@ max). The defaults live in `src/lib/validations/source-priority.ts:205-220`. Concrete consequences: -- If you already sync steps from a Withings ScanWatch and also import - Apple Health, the Apple-Health-aggregated stream wins for the day — - iOS HealthKit folds the watch sensor + iPhone motion data into a - single canonical stream, so it is the most complete cumulative - source. The Withings rows stay in the database as an audit trail - but drop out of the per-day aggregation. +- Native HealthKit cumulative statistics are the canonical Apple + Health stream for a day and outrank both XML estimates and Withings + cumulative rows. An XML upload chooses the largest source subtotal + only; it never claims to reproduce HealthKit's device de-duplication. + Lower-priority rows stay in the database as an audit trail but drop + out of the displayed per-day aggregation. - If you weigh yourself on a Withings scale and also import Apple Health (which received the same reading via the Health Mate iOS app), the Withings row wins for display. Apple Health is diff --git a/messages/de.json b/messages/de.json index 4dd6ca7fe..4946f88a0 100644 --- a/messages/de.json +++ b/messages/de.json @@ -4519,6 +4519,7 @@ "failed": "Der Import ist fehlgeschlagen.", "rowsImported": "Bisher {count} Einträge importiert", "doneSummary": "Fertig — {imported} von {read} Datensätzen importiert, {skipped} klinische Datensätze übersprungen.", + "estimateWarning": "Kumulative Summen für {count} Tag(e) wurden anhand der größten Exportquelle geschätzt. Native Apple-Health-Statistiken bleiben maßgeblich.", "phase": { "queued": "In Warteschlange…", "unpacking": "Archiv wird entpackt…", diff --git a/messages/en.json b/messages/en.json index c65c37f41..d93c8af70 100644 --- a/messages/en.json +++ b/messages/en.json @@ -4519,6 +4519,7 @@ "failed": "The import failed.", "rowsImported": "{count} rows imported so far", "doneSummary": "Done — {imported} of {read} records imported, {skipped} clinical records skipped.", + "estimateWarning": "Cumulative totals for {count} day(s) were estimated from the largest export source. Native Apple Health statistics remain authoritative.", "phase": { "queued": "Queued…", "unpacking": "Unpacking the archive…", diff --git a/messages/es.json b/messages/es.json index fac747d68..a97dd7895 100644 --- a/messages/es.json +++ b/messages/es.json @@ -4519,6 +4519,7 @@ "failed": "La importación falló.", "rowsImported": "{count} filas importadas hasta ahora", "doneSummary": "Listo: {imported} de {read} registros importados, {skipped} registros clínicos omitidos.", + "estimateWarning": "Los totales acumulados de {count} día(s) se estimaron a partir de la mayor fuente de la exportación. Las estadísticas nativas de Apple Health siguen siendo la referencia.", "phase": { "queued": "En cola…", "unpacking": "Descomprimiendo el archivo…", diff --git a/messages/fr.json b/messages/fr.json index 16648caaa..e8b8ad01f 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -4519,6 +4519,7 @@ "failed": "L'import a échoué.", "rowsImported": "{count} lignes importées jusqu'ici", "doneSummary": "Terminé — {imported} enregistrements sur {read} importés, {skipped} enregistrements cliniques ignorés.", + "estimateWarning": "Les totaux cumulés de {count} jour(s) ont été estimés à partir de la source d’export la plus élevée. Les statistiques Apple Santé natives restent la référence.", "phase": { "queued": "En file d'attente…", "unpacking": "Décompression de l'archive…", diff --git a/messages/it.json b/messages/it.json index 018c078eb..7334e0c7c 100644 --- a/messages/it.json +++ b/messages/it.json @@ -4519,6 +4519,7 @@ "failed": "L'importazione non è riuscita.", "rowsImported": "{count} righe importate finora", "doneSummary": "Fatto: {imported} di {read} record importati, {skipped} record clinici saltati.", + "estimateWarning": "I totali cumulativi di {count} giorno/i sono stati stimati dalla fonte di esportazione più alta. Le statistiche native di Apple Salute restano autorevoli.", "phase": { "queued": "In coda…", "unpacking": "Estrazione dell'archivio…", diff --git a/messages/pl.json b/messages/pl.json index e0c541e04..26b649f7f 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -4519,6 +4519,7 @@ "failed": "Import nie powiódł się.", "rowsImported": "Zaimportowano dotąd {count} wierszy", "doneSummary": "Gotowe — zaimportowano {imported} z {read} rekordów, pominięto {skipped} rekordów klinicznych.", + "estimateWarning": "Łączne wartości dla {count} dni oszacowano na podstawie największego źródła w eksporcie. Natywne statystyki Apple Health pozostają nadrzędne.", "phase": { "queued": "W kolejce…", "unpacking": "Rozpakowywanie archiwum…", diff --git a/package.json b/package.json index 3f89a16da..83afa99f9 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.31.6", + "version": "1.31.7", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/prisma/migrations/0263_apple_health_aggregate_authority/migration.sql b/prisma/migrations/0263_apple_health_aggregate_authority/migration.sql new file mode 100644 index 000000000..d98a8d979 --- /dev/null +++ b/prisma/migrations/0263_apple_health_aggregate_authority/migration.sql @@ -0,0 +1,27 @@ +-- Aggregate provenance makes the shared Apple Health `stats:*` identity +-- authority-aware. Existing rows cannot be classified safely from their stored +-- shape, so they remain repairable but are never promoted heuristically. +CREATE TYPE "measurement_aggregation_provenance" AS ENUM ( + 'HEALTHKIT_STATISTICS', + 'EXPORT_XML_SOURCE_MAX', + 'LEGACY_UNKNOWN' +); + +ALTER TABLE "measurements" + ADD COLUMN "aggregation_provenance" "measurement_aggregation_provenance", + ADD COLUMN "aggregation_contributor_count" INTEGER, + ADD COLUMN "aggregation_selected_source_hash" TEXT; + +UPDATE "measurements" +SET "aggregation_provenance" = 'LEGACY_UNKNOWN' +WHERE "source" = 'APPLE_HEALTH' + AND "external_id" LIKE 'stats:%'; + +-- Revision 1 denotes rows created under the pre-authority parser. New binaries +-- write revision 2 explicitly; the default stays legacy-safe during rollouts. +ALTER TABLE "import_jobs" + ADD COLUMN "parser_revision" INTEGER NOT NULL DEFAULT 1; + +UPDATE "import_jobs" +SET "parser_revision" = 1; + diff --git a/prisma/schema.prisma b/prisma/schema.prisma index e12e4d387..1fe1a326c 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1592,6 +1592,14 @@ enum SleepStage { @@map("sleep_stage") } +enum MeasurementAggregationProvenance { + HEALTHKIT_STATISTICS + EXPORT_XML_SOURCE_MAX + LEGACY_UNKNOWN + + @@map("measurement_aggregation_provenance") +} + model Measurement { id String @id @default(cuid()) userId String @map("user_id") @@ -1625,6 +1633,18 @@ model Measurement { // sample's `metadata.HKWasUserEntered`, Withings model code, …). // Stored as a string so the ingest path can replay it back unchanged. externalSourceVersion String? @map("external_source_version") + // Authority of an aggregate stored under a `stats:*` identity. NULL for + // ordinary point/sample rows. Native HealthKit statistics outrank the + // source-day estimate derived from export.xml; migrated ambiguous rows are + // explicitly LEGACY_UNKNOWN until a native sync or archive reprocess repairs + // them. + aggregationProvenance MeasurementAggregationProvenance? @map("aggregation_provenance") + // Distinct source identities contributing to an export.xml source-day + // estimate. Source labels themselves are never persisted. + aggregationContributorCount Int? @map("aggregation_contributor_count") + // SHA-256 of the deterministically selected source identity. Non-identifying + // provenance for repeatability without retaining device/source labels. + aggregationSelectedSourceHash String? @map("aggregation_selected_source_hash") // Only set for BLOOD_GLUCOSE measurements. NULL for all others. glucoseContext GlucoseContext? @map("glucose_context") // v1.4.23 — per-stage sleep label. NULL for non-sleep rows; non-NULL @@ -4284,6 +4304,9 @@ model ImportJob { /// kick-off endpoint short-circuit a re-upload of the exact same /// file back to the existing job id without re-queueing. uploadSha256 String? @map("upload_sha256") + /// Parser semantics revision. Existing archives migrated at revision 1 can + /// be deliberately reprocessed when the current parser revision advances. + parserRevision Int @default(1) @map("parser_revision") /// Apple's `ExportDate` attribute string-formatted, parsed during /// the unpack phase. NULL until the parser sees it. exportedAt DateTime? @map("exported_at") diff --git a/public/sw.js b/public/sw.js index 43e7d0f45..0718f9aa0 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.31.6"; + /* @sw-version-fallback */ "v1.31.7"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/__tests__/i18n-reverse-coverage.test.ts b/src/__tests__/i18n-reverse-coverage.test.ts index 5f2f3f5b6..438d20bca 100644 --- a/src/__tests__/i18n-reverse-coverage.test.ts +++ b/src/__tests__/i18n-reverse-coverage.test.ts @@ -245,5 +245,5 @@ describe("i18n reverse coverage", () => { "Wire a call site, or add a DB-driven prefix to DYNAMIC_ALLOWLIST_PREFIXES.", ); } - }); + }, 15_000); }); diff --git a/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts b/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts index 871b6fc77..9e505e322 100644 --- a/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts +++ b/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts @@ -188,9 +188,29 @@ describe("POST /api/admin/import-apple-health-export", () => { expect(body.data.jobId).not.toBe("old-failed-job"); expect(mocks.bossSend).toHaveBeenCalledTimes(1); expect(mocks.importJobCreate).toHaveBeenCalledTimes(1); + expect(mocks.importJobCreate).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ parserRevision: 2 }), + }), + ); + expect(mocks.bossSend).toHaveBeenCalledWith( + "apple-health-import-v2", + expect.objectContaining({ + userId: "user-2", + triggeredByAdminId: "admin-1", + }), + expect.objectContaining({ retryLimit: 0 }), + ); + expect(mocks.importJobUpdate).toHaveBeenCalledWith({ + where: { id: "ij-1" }, + data: { pgBossJobId: "boss-job-1" }, + }); expect(mocks.importJobFindFirst).toHaveBeenCalledWith( expect.objectContaining({ - where: expect.objectContaining({ status: { not: "failed" } }), + where: expect.objectContaining({ + status: { not: "failed" }, + parserRevision: 2, + }), }), ); expect(mocks.unlink).not.toHaveBeenCalled(); @@ -211,5 +231,10 @@ describe("POST /api/admin/import-apple-health-export", () => { expect(mocks.bossSend).not.toHaveBeenCalled(); expect(mocks.importJobCreate).not.toHaveBeenCalled(); expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH); + expect(mocks.importJobFindFirst).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ parserRevision: 2 }), + }), + ); }); }); diff --git a/src/app/api/admin/import-apple-health-export/route.ts b/src/app/api/admin/import-apple-health-export/route.ts index e19739d2d..67e8ecd32 100644 --- a/src/app/api/admin/import-apple-health-export/route.ts +++ b/src/app/api/admin/import-apple-health-export/route.ts @@ -21,7 +21,8 @@ import { auditLog } from "@/lib/auth/audit"; import { checkRateLimit } from "@/lib/rate-limit"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, + APPLE_HEALTH_IMPORT_PARSER_REVISION, APPLE_HEALTH_IMPORT_SEND_OPTIONS, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; @@ -121,6 +122,7 @@ export const POST = apiHandler(async (request: NextRequest) => { where: { userId: targetUser.id, uploadSha256: uploaded.sha256, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, status: { not: "failed" }, }, orderBy: { startedAt: "desc" }, @@ -157,6 +159,7 @@ export const POST = apiHandler(async (request: NextRequest) => { status: "queued", uploadBytes: uploaded.bytes, uploadSha256: uploaded.sha256, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); @@ -171,7 +174,7 @@ export const POST = apiHandler(async (request: NextRequest) => { // first run, so a redelivery could only fail on the deleted `/tmp` // file and mask the real outcome (see the worker's send-options doc). const bossJobId = await boss.send( - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, payload, APPLE_HEALTH_IMPORT_SEND_OPTIONS, ); diff --git a/src/app/api/import/apple-health-export/__tests__/route.test.ts b/src/app/api/import/apple-health-export/__tests__/route.test.ts index 09fe09aa6..557acb9c1 100644 --- a/src/app/api/import/apple-health-export/__tests__/route.test.ts +++ b/src/app/api/import/apple-health-export/__tests__/route.test.ts @@ -170,10 +170,27 @@ describe("POST /api/import/apple-health-export — dedup by content hash (issue expect(body.data.jobId).not.toBe("old-failed-job"); expect(mocks.bossSend).toHaveBeenCalledTimes(1); expect(mocks.importJobCreate).toHaveBeenCalledTimes(1); + expect(mocks.importJobCreate).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ parserRevision: 2 }), + }), + ); + expect(mocks.bossSend).toHaveBeenCalledWith( + "apple-health-import-v2", + expect.objectContaining({ userId: "user-1" }), + expect.objectContaining({ retryLimit: 0 }), + ); + expect(mocks.importJobUpdate).toHaveBeenCalledWith({ + where: { id: "ij-1" }, + data: { pgBossJobId: "boss-job-1" }, + }); // The dedup lookup must exclude failed rows. expect(mocks.importJobFindFirst).toHaveBeenCalledWith( expect.objectContaining({ - where: expect.objectContaining({ status: { not: "failed" } }), + where: expect.objectContaining({ + status: { not: "failed" }, + parserRevision: 2, + }), }), ); // A fresh job was staged — nothing to unlink on this path. @@ -196,5 +213,10 @@ describe("POST /api/import/apple-health-export — dedup by content hash (issue expect(mocks.importJobCreate).not.toHaveBeenCalled(); // The freshly staged upload is redundant and must be cleaned up. expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH); + expect(mocks.importJobFindFirst).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ parserRevision: 2 }), + }), + ); }); }); diff --git a/src/app/api/import/apple-health-export/route.ts b/src/app/api/import/apple-health-export/route.ts index fcd505d03..131fd2b68 100644 --- a/src/app/api/import/apple-health-export/route.ts +++ b/src/app/api/import/apple-health-export/route.ts @@ -4,7 +4,7 @@ * * Streams the upload directly to disk (`/tmp/healthlog-upload-*.bin`) * without buffering the full body in memory, creates an `ImportJob` - * row in `queued`, and enqueues an `apple-health-import` pg-boss job. + * row in `queued`, and enqueues an `apple-health-import-v2` pg-boss job. * Returns `{ jobId }` so the client can poll * `GET /api/import/apple-health-export/[jobId]/status`. * @@ -29,7 +29,8 @@ import { auditLog } from "@/lib/auth/audit"; import { checkRateLimit } from "@/lib/rate-limit"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, + APPLE_HEALTH_IMPORT_PARSER_REVISION, APPLE_HEALTH_IMPORT_SEND_OPTIONS, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; @@ -118,6 +119,7 @@ export const POST = apiHandler(async (request: NextRequest) => { where: { userId: user.id, uploadSha256: uploaded.sha256, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, status: { not: "failed" }, }, orderBy: { startedAt: "desc" }, @@ -154,6 +156,7 @@ export const POST = apiHandler(async (request: NextRequest) => { status: "queued", uploadBytes: uploaded.bytes, uploadSha256: uploaded.sha256, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); @@ -167,7 +170,7 @@ export const POST = apiHandler(async (request: NextRequest) => { // first run, so a redelivery could only fail on the deleted `/tmp` // file and mask the real outcome (see the worker's send-options doc). const bossJobId = await boss.send( - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, payload, APPLE_HEALTH_IMPORT_SEND_OPTIONS, ); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 1001392ce..8d229c2fb 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -516,6 +516,9 @@ async function postBatch(request: NextRequest): Promise { rhythmClassification: p.row .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"], deviceType: p.row.deviceType as string | null, + aggregationProvenance: statsRow ? "HEALTHKIT_STATISTICS" : null, + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, }, { exactExternalMatch: statsRow ? "update" : "duplicate" }, ), diff --git a/src/components/settings/__tests__/import-panel.test.tsx b/src/components/settings/__tests__/import-panel.test.tsx index 4448f86cb..8b55cb2d3 100644 --- a/src/components/settings/__tests__/import-panel.test.tsx +++ b/src/components/settings/__tests__/import-panel.test.tsx @@ -26,6 +26,8 @@ vi.mock("next/navigation", () => ({ usePathname: () => "/settings/export", })); +import type { Locale } from "@/lib/i18n/config"; +import { AppleHealthEstimateWarning } from "../import-panel/apple-health-import-card"; import { I18nProvider } from "@/lib/i18n/context"; import { ImportPanel, @@ -33,9 +35,10 @@ import { EXAMPLE_CSV, parseImportJson, } from "../import-panel"; + import { parseCsvMeasurements } from "@/lib/import/csv-measurements"; -function render(node: React.ReactElement, locale: "en" | "de" = "en") { +function render(node: React.ReactElement, locale: Locale = "en") { const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false, staleTime: Infinity } }, }); @@ -98,6 +101,20 @@ describe(" — SSR smoke", () => { }); }); +describe("", () => { + it.each(["en", "de", "es", "fr", "it", "pl"] as const)( + "renders localized estimate disclosure for %s", + (locale) => { + const html = render(, locale); + expect(html).toContain('data-testid="apple-health-estimate-warning"'); + expect(html).toContain("3"); + expect(html).not.toContain( + "settings.sections.export.import.appleHealth.estimateWarning", + ); + }, + ); +}); + describe("EXAMPLE_IMPORT payload", () => { it("round-trips through JSON and stays parseable", () => { const serialised = JSON.stringify(EXAMPLE_IMPORT, null, 2); diff --git a/src/components/settings/import-panel/apple-health-import-card.tsx b/src/components/settings/import-panel/apple-health-import-card.tsx index 1ea703256..ce22accec 100644 --- a/src/components/settings/import-panel/apple-health-import-card.tsx +++ b/src/components/settings/import-panel/apple-health-import-card.tsx @@ -27,10 +27,31 @@ interface JobStatus { result: { totals?: { recordsRead?: number; rowsUpserted?: number }; clinical?: { skipped?: number }; + cumulativeEstimates?: { days?: number; rows?: number }; } | null; failureReason: string | null; } +export function AppleHealthEstimateWarning({ days }: { days: number }) { + const { t } = useTranslations(); + return ( +

+

+ ); +} + export function AppleHealthImportCard() { const { t } = useTranslations(); const fileInputRef = useRef(null); @@ -233,6 +254,11 @@ export function AppleHealthImportCard() { )} + {isDone && (status?.result?.cumulativeEstimates?.days ?? 0) > 0 && ( + + )} {(isFailed || uploadError) && (

{ expect(body.response_format).toEqual({ type: "json_object" }); // No seed passed → field omitted from the body entirely. expect(body).not.toHaveProperty("seed"); + expect(body.max_tokens).toBe(1000); + expect(body.temperature).toBe(0.3); + expect(body).not.toHaveProperty("max_completion_tokens"); }); it("forces JSON mode only when responseFormat is json AND no tools (M-1)", async () => { @@ -147,6 +150,118 @@ describe("OpenAIClient", () => { expect(body.seed).toBe(1234); }); + it.each([ + ["gpt-5.4-nano", 1000, "https://api.openai.com/v1"], + ["gpt-5-2025-08-07", 321, "https://api.openai.com/v1"], + ["o1", 321, "https://api.openai.com/v1"], + ["o3-mini", 321, "https://api.openai.com/v1"], + ["o4-mini-2025-04-16", 321, "https://api.openai.com/v1"], + ["gpt-5.4-mini", 321, "https://API.OPENAI.COM:443/v1/"], + ])( + "uses the canonical modern token wire for %s", + async (model, expectedBudget, baseUrl) => { + const mockFetch = vi.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + choices: [{ message: { content: "ok" } }], + usage: { total_tokens: 5 }, + }), + }); + vi.stubGlobal("fetch", mockFetch); + + const client = new OpenAIClient({ + apiKey: "sk-test", + model, + baseUrl, + }); + + await client.generateCompletion( + singleUserTurn({ + system: "test", + user: "test", + temperature: 0.7, + ...(expectedBudget === 1000 ? {} : { maxTokens: expectedBudget }), + seed: 1234, + }), + ); + + const body = JSON.parse(mockFetch.mock.calls[0][1].body); + expect(body.max_completion_tokens).toBe(expectedBudget); + expect(body).not.toHaveProperty("max_tokens"); + expect(body).not.toHaveProperty("temperature"); + expect(body).not.toHaveProperty("seed"); + }, + ); + + it("keeps the legacy sampling wire for canonical GPT-4 models", async () => { + const mockFetch = vi.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + choices: [{ message: { content: "ok" } }], + usage: { total_tokens: 5 }, + }), + }); + vi.stubGlobal("fetch", mockFetch); + + const client = new OpenAIClient({ + apiKey: "sk-test", + model: "gpt-4o-mini", + baseUrl: "https://api.openai.com/v1", + }); + + await client.generateCompletion( + singleUserTurn({ + system: "test", + user: "test", + temperature: 0.7, + maxTokens: 321, + seed: 1234, + }), + ); + + const body = JSON.parse(mockFetch.mock.calls[0][1].body); + expect(body.max_tokens).toBe(321); + expect(body.temperature).toBe(0.7); + expect(body.seed).toBe(1234); + expect(body).not.toHaveProperty("max_completion_tokens"); + }); + + it("keeps the legacy sampling wire for GPT-5-named models on compatible gateways", async () => { + const mockFetch = vi.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + choices: [{ message: { content: "ok" } }], + usage: { total_tokens: 5 }, + }), + }); + vi.stubGlobal("fetch", mockFetch); + + const client = new OpenAIClient({ + apiKey: "sk-test", + model: "gpt-5.4-nano", + baseUrl: "https://gateway.example.com/v1", + }); + + await client.generateCompletion( + singleUserTurn({ + system: "test", + user: "test", + temperature: 0.7, + maxTokens: 321, + seed: 1234, + }), + ); + + const body = JSON.parse(mockFetch.mock.calls[0][1].body); + expect(body.max_tokens).toBe(321); + expect(body.temperature).toBe(0.7); + expect(body.seed).toBe(1234); + expect(body).not.toHaveProperty("max_completion_tokens"); + }); + it("throws on non-ok response", async () => { vi.stubGlobal( "fetch", diff --git a/src/lib/ai/openai-capabilities.ts b/src/lib/ai/openai-capabilities.ts new file mode 100644 index 000000000..75b723a8e --- /dev/null +++ b/src/lib/ai/openai-capabilities.ts @@ -0,0 +1,67 @@ +export type OpenAIChatCompletionsCapabilities = + | { + tokenBudgetField: "max_completion_tokens"; + supportsSamplingControls: false; + } + | { + tokenBudgetField: "max_tokens"; + supportsSamplingControls: true; + }; + +const MODERN_CANONICAL_CAPABILITIES = { + tokenBudgetField: "max_completion_tokens", + supportsSamplingControls: false, +} as const; + +const LEGACY_COMPATIBLE_CAPABILITIES = { + tokenBudgetField: "max_tokens", + supportsSamplingControls: true, +} as const; + +function isModernCanonicalModel(model: string): boolean { + const normalizedModel = model.toLowerCase(); + + return ( + normalizedModel === "gpt-5" || + normalizedModel.startsWith("gpt-5-") || + normalizedModel.startsWith("gpt-5.") || + normalizedModel === "o1" || + normalizedModel.startsWith("o1-") || + normalizedModel === "o3" || + normalizedModel.startsWith("o3-") || + normalizedModel === "o4" || + normalizedModel.startsWith("o4-") + ); +} + +function isCanonicalOpenAIEndpoint(baseUrl: string): boolean { + try { + const url = new URL(baseUrl); + return ( + url.protocol === "https:" && + url.hostname === "api.openai.com" && + url.port === "" && + url.pathname.replace(/\/+$/, "") === "/v1" && + url.username === "" && + url.password === "" && + url.search === "" && + url.hash === "" + ); + } catch { + return false; + } +} + +/** + * Select Chat Completions request capabilities from both endpoint and model. + * Custom OpenAI-compatible gateways retain the broadly supported legacy wire, + * even when their model name resembles a modern OpenAI model. + */ +export function selectOpenAIChatCompletionsCapabilities( + baseUrl: string, + model: string, +): OpenAIChatCompletionsCapabilities { + return isCanonicalOpenAIEndpoint(baseUrl) && isModernCanonicalModel(model) + ? MODERN_CANONICAL_CAPABILITIES + : LEGACY_COMPATIBLE_CAPABILITIES; +} diff --git a/src/lib/ai/openai-client.ts b/src/lib/ai/openai-client.ts index 510613830..eb1e140ce 100644 --- a/src/lib/ai/openai-client.ts +++ b/src/lib/ai/openai-client.ts @@ -1,5 +1,6 @@ import { safeFetch } from "@/lib/safe-fetch"; import type { AIProvider, CompletionParams, CompletionResult } from "./types"; +import { selectOpenAIChatCompletionsCapabilities } from "./openai-capabilities"; import { buildOpenAIMessages, buildOpenAITools, @@ -60,6 +61,18 @@ export class OpenAIClient implements AIProvider { // therefore stay out of JSON mode, and every tool round is non-JSON by // construction. Insight/extraction callers opt in with `responseFormat:"json"`. const useJsonFormat = params.responseFormat === "json" && !hasTools; + const capabilities = selectOpenAIChatCompletionsCapabilities( + this.config.baseUrl, + this.config.model, + ); + const tokenBudget = params.maxTokens ?? 1000; + const capabilityParams = capabilities.supportsSamplingControls + ? { + [capabilities.tokenBudgetField]: tokenBudget, + temperature: params.temperature ?? 0.3, + ...(params.seed !== undefined ? { seed: params.seed } : {}), + } + : { [capabilities.tokenBudgetField]: tokenBudget }; const res = await safeFetch( url, @@ -72,15 +85,10 @@ export class OpenAIClient implements AIProvider { body: JSON.stringify({ model: this.config.model, messages, - temperature: params.temperature ?? 0.3, - max_tokens: params.maxTokens ?? 1000, + ...capabilityParams, ...(useJsonFormat ? { response_format: { type: "json_object" } } : {}), - // Deterministic seed for reproducible reference output; omitted - // (undefined → dropped by JSON.stringify) when the caller does - // not pin one (e.g. the seedless daily-briefing re-roll). - ...(params.seed !== undefined ? { seed: params.seed } : {}), ...(tools ? { tools } : {}), ...(params.toolChoice ? { tool_choice: params.toolChoice } : {}), }), diff --git a/src/lib/daily/__tests__/digest.test.ts b/src/lib/daily/__tests__/digest.test.ts index 4cc3fd4e1..a8214d9e0 100644 --- a/src/lib/daily/__tests__/digest.test.ts +++ b/src/lib/daily/__tests__/digest.test.ts @@ -236,6 +236,49 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => { expect(dose?.actions[0].href).toBe("/medications"); }); + it("emits one medication-specific card and link per actionable candidate", () => { + const d = buildDailyDigest( + input({ + medsToday: meds({ + activeCount: 2, + dueCandidates: [ + { + medicationId: "med-ramipril", + medicationName: "Ramipril", + dueAt: new Date(NOW.getTime() - 30 * 60_000).toISOString(), + overdue: true, + availableFrom: new Date( + NOW.getTime() - 2 * 60 * 60_000, + ).toISOString(), + }, + { + medicationId: "med-mounjaro", + medicationName: "Mounjaro", + dueAt: new Date(NOW.getTime() - 30 * 60_000).toISOString(), + overdue: false, + availableFrom: new Date( + NOW.getTime() - 24 * 60 * 60_000, + ).toISOString(), + }, + ], + }), + }), + t, + ); + + const doses = d.worthALook.filter((item) => item.kind === "dose_window"); + expect(doses).toHaveLength(2); + expect(doses.map((item) => item.status)).toEqual(["warning", "info"]); + expect(doses.map((item) => item.body)).toEqual([ + t("daily.item.doseWindow.overdueBodyNamed", { name: "Ramipril" }), + t("daily.item.doseWindow.bodyNamed", { name: "Mounjaro" }), + ]); + expect(doses.map((item) => item.actions[0]?.href)).toEqual([ + "/medications?highlight=med-ramipril", + "/medications?highlight=med-mounjaro", + ]); + }); + it("does NOT emit a dose-window item when the medications module is off", () => { const d = buildDailyDigest( input({ @@ -334,6 +377,58 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => { expect(d.worthALook.some((i) => i.kind === "dose_window")).toBe(false); }); + it("shows a weekly future slot once its engine-derived availability window opens", () => { + const d = buildDailyDigest( + input({ + medsToday: meds({ + dueCandidates: [ + { + medicationId: "med-weekly", + medicationName: "Weekly dose", + dueAt: new Date(NOW.getTime() + 24 * 60 * 60_000).toISOString(), + overdue: false, + availableFrom: new Date(NOW.getTime() - 60_000).toISOString(), + }, + ], + }), + }), + t, + ); + + expect( + d.worthALook.filter((item) => item.kind === "dose_window"), + ).toHaveLength(1); + }); + + it("respects an explicit weekly availability boundary exactly", () => { + const candidate = { + medicationId: "med-weekly", + medicationName: "Weekly dose", + dueAt: new Date(NOW.getTime() + 24 * 60 * 60_000).toISOString(), + overdue: false, + availableFrom: NOW.toISOString(), + }; + + const atBoundary = buildDailyDigest( + input({ medsToday: meds({ dueCandidates: [candidate] }) }), + t, + ); + const beforeBoundary = buildDailyDigest( + input({ + now: new Date(NOW.getTime() - 1), + medsToday: meds({ dueCandidates: [candidate] }), + }), + t, + ); + + expect( + atBoundary.worthALook.some((item) => item.kind === "dose_window"), + ).toBe(true); + expect( + beforeBoundary.worthALook.some((item) => item.kind === "dose_window"), + ).toBe(false); + }); + it("keeps a dose beyond the takeable lead off the rail", () => { const d = buildDailyDigest( input({ @@ -350,9 +445,7 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => { expect(d.worthALook.some((i) => i.kind === "dose_window")).toBe(false); }); - it("renders neither face for a cached block whose anchor has since passed", () => { - // `MedsTodayBlock` contract: a past `nextDueAt` with `nextDueOverdue: - // false` means the anchor passed AFTER the block was built. + it("keeps a stale cached non-overdue scalar calm instead of dropping or escalating it", () => { const d = buildDailyDigest( input({ medsToday: meds({ @@ -363,7 +456,11 @@ describe("buildDailyDigest — worth-a-look rail item builders", () => { }), t, ); - expect(d.worthALook.some((i) => i.kind === "dose_window")).toBe(false); + const dose = d.worthALook.find((item) => item.kind === "dose_window"); + expect(dose?.status).toBe("info"); + expect(dose?.body).toBe( + t("daily.item.doseWindow.bodyNamed", { name: "Ramipril" }), + ); }); it("maps one sync-issue item per broken integration", () => { diff --git a/src/lib/daily/digest.ts b/src/lib/daily/digest.ts index d38e0453b..063e4ad24 100644 --- a/src/lib/daily/digest.ts +++ b/src/lib/daily/digest.ts @@ -320,76 +320,85 @@ export const DOSE_DUE_LOOKAHEAD_MS = 60_000; /** - * Dose-window item — the day's medication action, in two distinct faces. + * Compose one card for every medication candidate the canonical scheduling + * engine says is currently available. The additive collection is authoritative + * when present; scalar fields remain the fallback for older cached snapshots. * - * OVERDUE: an open slot whose anchor has passed. `warning` status, past-due - * wording — the one genuinely time-critical daily action. - * - * DUE: a slot the user can already take (within `DOSE_DUE_LOOKAHEAD_MS`) but - * that is not yet late. `info` status, "due soon" wording — present on the - * card because Today is a surface the user OPENS to see what the day asks of - * them, not a notification that arrives uninvited. The no-nagging rule governs - * PUSH (the reminder cron still fires only on overdue); a deliberately-opened - * card that stays silent about a dose the user can take right now is simply - * withholding the answer the surface exists to give. - * - * Neither face is manufactured on a settled day: once every scheduled dose is - * resolved, the next display-due slot lies beyond the lookahead and nothing is - * emitted. A `nextDueAt` in the PAST with `nextDueOverdue: false` is the - * documented cached-block shape (`MedsTodayBlock`) — the anchor passed after - * the block was built — and renders as neither face until a fresh block lands. - * - * Gated on the `medications` module; carries a single log-dose action. + * `overdue` is never inferred from the wall clock. A cached non-overdue + * candidate whose anchor has passed stays calm/informational until refresh. */ -function buildDoseWindowItem( +function buildDoseWindowItems( meds: MedsTodayBlock, modules: DigestModuleMap, now: Date, t: Translate, -): PriorityItem | null { - if (!moduleEnabled(modules, "medications")) return null; +): PriorityItem[] { + if (!moduleEnabled(modules, "medications")) return []; - const name = meds.nextDueMedicationName; - const id = meds.nextDueMedicationId; - const action = { - labelKey: "daily.action.logDose", - intent: "dose.log", - // Deep-link straight to the due medication's card so the tap lands on the - // right one instead of a generic list the user then has to scan (the id is - // known server-side whenever the name is). The bare list stays the honest - // fallback for an older cached block that predates the id field. - href: id ? `/medications?highlight=${id}` : "/medications", - }; + const candidates = + meds.dueCandidates !== undefined + ? meds.dueCandidates + : meds.nextDueOverdue || meds.nextDueAt + ? [ + { + medicationId: meds.nextDueMedicationId, + medicationName: meds.nextDueMedicationName, + dueAt: meds.nextDueAt, + overdue: meds.nextDueOverdue, + availableFrom: null, + }, + ] + : []; + const items: PriorityItem[] = []; - if (meds.nextDueOverdue) { - return { - kind: "dose_window", - title: t("daily.item.doseWindow.overdueTitle"), - body: name - ? t("daily.item.doseWindow.overdueBodyNamed", { name }) - : t("daily.item.doseWindow.overdueBody"), - status: "warning", - actions: [action], - moduleKey: "medications", + for (const candidate of candidates) { + if (!candidate.overdue) { + const dueAt = candidate.dueAt ? Date.parse(candidate.dueAt) : NaN; + if (!Number.isFinite(dueAt)) continue; + + if (candidate.availableFrom !== null) { + const availableFrom = Date.parse(candidate.availableFrom); + if (!Number.isFinite(availableFrom) || now.getTime() < availableFrom) { + continue; + } + } else if (dueAt - now.getTime() > DOSE_DUE_LOOKAHEAD_MS) { + continue; + } + } + + const name = candidate.medicationName; + const id = candidate.medicationId; + const action = { + labelKey: "daily.action.logDose", + intent: "dose.log", + href: id ? `/medications?highlight=${id}` : "/medications", }; + items.push( + candidate.overdue + ? { + kind: "dose_window", + title: t("daily.item.doseWindow.overdueTitle"), + body: name + ? t("daily.item.doseWindow.overdueBodyNamed", { name }) + : t("daily.item.doseWindow.overdueBody"), + status: "warning", + actions: [action], + moduleKey: "medications", + } + : { + kind: "dose_window", + title: t("daily.item.doseWindow.title"), + body: name + ? t("daily.item.doseWindow.bodyNamed", { name }) + : t("daily.item.doseWindow.body"), + status: "info", + actions: [action], + moduleKey: "medications", + }, + ); } - if (!meds.nextDueAt) return null; - const dueAt = Date.parse(meds.nextDueAt); - if (!Number.isFinite(dueAt)) return null; - const lead = dueAt - now.getTime(); - if (lead < 0 || lead > DOSE_DUE_LOOKAHEAD_MS) return null; - - return { - kind: "dose_window", - title: t("daily.item.doseWindow.title"), - body: name - ? t("daily.item.doseWindow.bodyNamed", { name }) - : t("daily.item.doseWindow.body"), - status: "info", - actions: [action], - moduleKey: "medications", - }; + return items; } /** One rail item per broken integration — reconnect to keep data current. */ @@ -735,13 +744,9 @@ export function buildDailyDigest( // check-up least urgent. Bounded to 3 — a check-in the cap crowds out // resurfaces on a following day (within its window), never lost. const worthALook: PriorityItem[] = []; - const dose = buildDoseWindowItem( - input.medsToday, - input.modules, - input.now, - t, + worthALook.push( + ...buildDoseWindowItems(input.medsToday, input.modules, input.now, t), ); - if (dose) worthALook.push(dose); // A freshly-reached milestone is rare and one-per-day — surface it ahead of // the ambient sync / check-in items so the calm reward is not buried, but // below an overdue dose (the one genuinely time-critical daily action). diff --git a/src/lib/dashboard/__tests__/meds-today.test.ts b/src/lib/dashboard/__tests__/meds-today.test.ts index 4c96d84d6..5b466f0b9 100644 --- a/src/lib/dashboard/__tests__/meds-today.test.ts +++ b/src/lib/dashboard/__tests__/meds-today.test.ts @@ -208,6 +208,7 @@ describe("buildMedsTodayBlock — tally", () => { nextDueOverdue: false, nextDueMedicationName: null, nextDueMedicationId: null, + dueCandidates: [], }); }); @@ -317,6 +318,63 @@ describe("buildMedsTodayBlock — next due (real engine)", () => { expect(block.nextDueOverdue).toBe(false); }); + it("preserves equal-time medications and orders the overdue candidate first", async () => { + medications = [ + medication({ + id: "med-mounjaro", + name: "Mounjaro", + startsOn: new Date("2026-06-10T00:00:00.000Z"), + schedules: [ + dailySchedule({ + id: "sched-mounjaro", + medicationId: "med-mounjaro", + windowStart: "11:00", + windowEnd: "11:00", + timesOfDay: ["11:00"], + rrule: null, + rollingIntervalDays: 7, + }), + ], + }), + medication({ + id: "med-ramipril", + name: "Ramipril", + schedules: [ + dailySchedule({ + id: "sched-ramipril", + medicationId: "med-ramipril", + windowStart: "11:00", + windowEnd: "11:00", + timesOfDay: ["11:00"], + }), + ], + }), + ]; + + const block = await buildMedsTodayBlock(fakePrisma, "user-1", BERLIN, NOW); + + expect(block.dueCandidates).toEqual([ + { + medicationId: "med-ramipril", + medicationName: "Ramipril", + dueAt: "2026-06-10T09:00:00.000Z", + overdue: true, + availableFrom: "2026-06-10T07:00:00.000Z", + }, + { + medicationId: "med-mounjaro", + medicationName: "Mounjaro", + dueAt: "2026-06-10T09:00:00.000Z", + overdue: false, + availableFrom: "2026-06-09T08:00:00.000Z", + }, + ]); + expect(block.nextDueAt).toBe("2026-06-10T09:00:00.000Z"); + expect(block.nextDueOverdue).toBe(true); + expect(block.nextDueMedicationName).toBe("Ramipril"); + expect(block.nextDueMedicationId).toBe("med-ramipril"); + }); + it("returns null next-due when no schedule has an upcoming slot", async () => { // Course ended last month — the engine yields nothing. medications = [ diff --git a/src/lib/dashboard/meds-today.ts b/src/lib/dashboard/meds-today.ts index d550a5b35..1790ae439 100644 --- a/src/lib/dashboard/meds-today.ts +++ b/src/lib/dashboard/meds-today.ts @@ -44,6 +44,15 @@ import { } from "@/lib/medications/scheduling/next-due"; import { getUserTodayBounds } from "@/lib/tz/local-day"; +export interface MedsTodayDueCandidate { + medicationId: string; + medicationName: string; + dueAt: string; + overdue: boolean; + /** Canonical attribution-band start for this slot (ISO8601). */ + availableFrom: string; +} + export interface MedsTodayBlock { /** Active medications (paused courses excluded). */ activeCount: number; @@ -75,6 +84,11 @@ export interface MedsTodayBlock { * list. */ nextDueMedicationId: string | null; + /** + * Every active medication's current display-due candidate. Optional only so + * older cached snapshots remain valid; the live builder always emits it. + */ + dueCandidates?: MedsTodayDueCandidate[]; } export async function buildMedsTodayBlock( @@ -183,13 +197,7 @@ export async function buildMedsTodayBlock( eraStartByMedId.set(f.medicationId, f._max.validUntil); } - // Earliest display-due across active medications. An open overdue - // slot's anchor is in the past, so the minimum naturally prefers it - // over any future slot. - let nextDueAt: Date | null = null; - let nextDueOverdue = false; - let nextDueMedicationName: string | null = null; - let nextDueMedicationId: string | null = null; + const dueCandidates: MedsTodayDueCandidate[] = []; for (const m of medications) { const display = computeDisplayDue({ medication: { @@ -207,22 +215,32 @@ export async function buildMedsTodayBlock( eraStart: eraStartByMedId.get(m.id) ?? null, }); if (!display) continue; - if (nextDueAt === null || display.at.getTime() < nextDueAt.getTime()) { - nextDueAt = display.at; - nextDueOverdue = display.overdue; - nextDueMedicationName = m.name; - nextDueMedicationId = m.id; - } + dueCandidates.push({ + medicationId: m.id, + medicationName: m.name, + dueAt: display.at.toISOString(), + overdue: display.overdue, + availableFrom: (display.availableFrom ?? display.at).toISOString(), + }); } + dueCandidates.sort( + (a, b) => + Number(b.overdue) - Number(a.overdue) || + a.availableFrom.localeCompare(b.availableFrom) || + a.dueAt.localeCompare(b.dueAt) || + a.medicationId.localeCompare(b.medicationId), + ); + const next = dueCandidates[0] ?? null; return { activeCount: medications.length, scheduledToday: todayEvents.length, takenToday, skippedToday, - nextDueAt: nextDueAt ? nextDueAt.toISOString() : null, - nextDueOverdue, - nextDueMedicationName, - nextDueMedicationId, + nextDueAt: next?.dueAt ?? null, + nextDueOverdue: next?.overdue ?? false, + nextDueMedicationName: next?.medicationName ?? null, + nextDueMedicationId: next?.medicationId ?? null, + dueCandidates, }; } diff --git a/src/lib/jobs/__tests__/apple-health-import-legacy-bridge.test.ts b/src/lib/jobs/__tests__/apple-health-import-legacy-bridge.test.ts new file mode 100644 index 000000000..1b50253ee --- /dev/null +++ b/src/lib/jobs/__tests__/apple-health-import-legacy-bridge.test.ts @@ -0,0 +1,92 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job } from "pg-boss"; + +const mocks = vi.hoisted(() => ({ + findUnique: vi.fn(), + update: vi.fn(), + send: vi.fn(), + getGlobalBoss: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + importJob: { + findUnique: mocks.findUnique, + update: mocks.update, + }, + }, + toJson: (value: unknown) => value, +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: mocks.getGlobalBoss, +})); + +import { + APPLE_HEALTH_IMPORT_PARSER_REVISION, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, + APPLE_HEALTH_IMPORT_V2_QUEUE, + migrateLegacyAppleHealthImport, + type AppleHealthImportPayload, +} from "../apple-health-import-worker"; + +const payload: AppleHealthImportPayload = { + userId: "user-1", + uploadPath: "/tmp/export.zip", + uploadBytes: 100, + enqueuedAt: "2026-07-21T10:00:00.000Z", +}; + +function legacyJob(): Job { + return { + id: "legacy-boss-1", + data: payload, + } as Job; +} + +beforeEach(() => { + vi.resetAllMocks(); + mocks.getGlobalBoss.mockReturnValue({ send: mocks.send }); + mocks.send.mockResolvedValue("v2-boss-1"); + mocks.findUnique.mockResolvedValue({ + id: "import-1", + status: "queued", + parserRevision: 1, + }); + mocks.update.mockResolvedValue({ id: "import-1" }); +}); + +describe("migrateLegacyAppleHealthImport", () => { + it("moves a legacy job onto the v2 queue while preserving its mirror id", async () => { + await migrateLegacyAppleHealthImport(legacyJob()); + + expect(mocks.send).toHaveBeenCalledWith( + APPLE_HEALTH_IMPORT_V2_QUEUE, + payload, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, + ); + expect(mocks.update).toHaveBeenCalledWith({ + where: { id: "import-1" }, + data: { + pgBossJobId: "v2-boss-1", + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + status: "queued", + failureReason: null, + completedAt: null, + }, + }); + }); + + it("does not migrate a terminal legacy mirror", async () => { + mocks.findUnique.mockResolvedValue({ + id: "import-1", + status: "done", + parserRevision: 1, + }); + + await migrateLegacyAppleHealthImport(legacyJob()); + + expect(mocks.send).not.toHaveBeenCalled(); + expect(mocks.update).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts b/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts index 64367ac29..5e5de5c6c 100644 --- a/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts +++ b/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts @@ -19,13 +19,38 @@ vi.mock("@/lib/db", () => ({ toJson: (value: unknown) => value, })); -import { APPLE_HEALTH_IMPORT_SEND_OPTIONS } from "../apple-health-import-worker"; +import { + APPLE_HEALTH_IMPORT_PARSER_REVISION, + APPLE_HEALTH_IMPORT_LEGACY_QUEUE, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, + APPLE_HEALTH_IMPORT_V2_QUEUE, +} from "../apple-health-import-worker"; const ROUTE_FILES = [ "src/app/api/import/apple-health-export/route.ts", "src/app/api/admin/import-apple-health-export/route.ts", ]; +const maintenanceSource = readFileSync( + join(process.cwd(), "src/lib/jobs/reminder/register-maintenance.ts"), + "utf8", +); +const workerSource = readFileSync( + join(process.cwd(), "src/lib/jobs/apple-health-import-worker.ts"), + "utf8", +); +const migrationSource = readFileSync( + join( + process.cwd(), + "prisma/migrations/0263_apple_health_aggregate_authority/migration.sql", + ), + "utf8", +); +const schemaSource = readFileSync( + join(process.cwd(), "prisma/schema.prisma"), + "utf8", +); + describe("apple health import — pg-boss send policy", () => { it("disables retries and widens the expiration window", () => { // retryLimit 0: the staged upload is unlinked by the first run, so @@ -42,8 +67,52 @@ describe("apple health import — pg-boss send policy", () => { for (const file of ROUTE_FILES) { const source = readFileSync(join(process.cwd(), file), "utf8"); expect(source).toMatch( - /boss\.send\(\s*APPLE_HEALTH_IMPORT_QUEUE,\s*payload,\s*APPLE_HEALTH_IMPORT_SEND_OPTIONS,?\s*\)/, + /boss\.send\(\s*APPLE_HEALTH_IMPORT_V2_QUEUE,\s*payload,\s*APPLE_HEALTH_IMPORT_SEND_OPTIONS,?\s*\)/, ); } }); }); + +describe("apple health import — parser revision boundary", () => { + it("isolates revision-2 sends from legacy workers", () => { + expect(APPLE_HEALTH_IMPORT_PARSER_REVISION).toBe(2); + expect(APPLE_HEALTH_IMPORT_V2_QUEUE).toBe("apple-health-import-v2"); + expect(APPLE_HEALTH_IMPORT_LEGACY_QUEUE).toBe("apple-health-import"); + }); + + it("drains revision-2 jobs separately while bridging legacy backlog", () => { + const allQueues = maintenanceSource.match( + /const allQueues\s*=\s*\[([\s\S]*?)\];/, + ); + expect(allQueues).not.toBeNull(); + expect(allQueues![1]).toMatch(/\bAPPLE_HEALTH_IMPORT_V2_QUEUE\b/); + expect(allQueues![1]).toMatch(/\bAPPLE_HEALTH_IMPORT_LEGACY_QUEUE\b/); + expect(maintenanceSource).toMatch( + /boss\.work[\s\S]{0,200}APPLE_HEALTH_IMPORT_V2_QUEUE[\s\S]{0,400}handleAppleHealthImport/, + ); + expect(maintenanceSource).toMatch( + /boss\.work[\s\S]{0,200}APPLE_HEALTH_IMPORT_LEGACY_QUEUE[\s\S]{0,400}migrateLegacyAppleHealthImport/, + ); + }); + + it("keeps the database default at revision 1 for legacy binaries", () => { + expect(migrationSource).toMatch( + /ADD COLUMN "parser_revision" INTEGER NOT NULL DEFAULT 1/, + ); + expect(migrationSource).not.toMatch( + /ALTER COLUMN "parser_revision" SET DEFAULT 2/, + ); + expect(schemaSource).toMatch( + /parserRevision\s+Int\s+@default\(1\)\s+@map\("parser_revision"\)/, + ); + }); + + it("explicitly creates revision-2 stand-ins and reconciles only revision-2 mirrors", () => { + expect(workerSource).toMatch( + /importJob\.create\([\s\S]{0,400}parserRevision:\s*APPLE_HEALTH_IMPORT_PARSER_REVISION/, + ); + expect(workerSource).toMatch( + /importJob\.findMany\(\{\s*where:\s*\{[\s\S]{0,200}parserRevision:\s*APPLE_HEALTH_IMPORT_PARSER_REVISION/, + ); + }); +}); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index e599c1e8a..662b144c2 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -1,5 +1,5 @@ /** - * v1.4.34 — pg-boss handler for the `apple-health-import` queue. + * v1.4.34 — pg-boss handler for the revision-2 Apple Health import queue. * * The kick-off endpoint (`POST /api/import/apple-health-export`, * `POST /api/admin/import-apple-health-export`) writes the upload to @@ -38,8 +38,14 @@ import { } from "@/lib/measurements/import-apple-health-export"; import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; -/** Queue name for the Apple Health export ingest. */ -export const APPLE_HEALTH_IMPORT_QUEUE = "apple-health-import"; +/** Queue name isolated from revision-1 workers during rolling deployments. */ +export const APPLE_HEALTH_IMPORT_V2_QUEUE = "apple-health-import-v2"; + +/** Queue used by pre-revision-2 binaries; new workers only bridge its backlog. */ +export const APPLE_HEALTH_IMPORT_LEGACY_QUEUE = "apple-health-import"; + +/** Parser semantics carried by every newly-created ImportJob. */ +export const APPLE_HEALTH_IMPORT_PARSER_REVISION = 2; /** Concurrency cap per worker host. */ export const APPLE_HEALTH_IMPORT_CONCURRENCY = 1; @@ -96,6 +102,58 @@ function getWorkerPrisma(): PrismaClient { return workerPrismaSingleton ?? prisma; } +/** + * Move a job claimed from the legacy queue onto the revision-2 queue without + * parsing it under the wrong revision. The existing ImportJob id remains the + * status handle; only its pg-boss mirror id and truthful parser revision move. + */ +export async function migrateLegacyAppleHealthImport( + job: Job, +): Promise { + const prisma = getWorkerPrisma(); + let importJob = await prisma.importJob.findUnique({ + where: { pgBossJobId: job.id }, + }); + if (!importJob) { + const { promise, resolve } = Promise.withResolvers(); + setTimeout(resolve, 250); + await promise; + importJob = await prisma.importJob.findUnique({ + where: { pgBossJobId: job.id }, + }); + } + if (!importJob) { + throw new Error(`No ImportJob mirror for legacy pg-boss job ${job.id}`); + } + if (importJob.status === "done" || importJob.status === "failed") { + return; + } + + const boss = getGlobalBoss(); + if (!boss) { + throw new Error("Cannot bridge legacy Apple Health import without pg-boss"); + } + const v2BossJobId = await boss.send( + APPLE_HEALTH_IMPORT_V2_QUEUE, + job.data, + APPLE_HEALTH_IMPORT_SEND_OPTIONS, + ); + if (!v2BossJobId) { + throw new Error("Failed to enqueue bridged Apple Health import"); + } + + await prisma.importJob.update({ + where: { id: importJob.id }, + data: { + pgBossJobId: v2BossJobId, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + status: "queued", + failureReason: null, + completedAt: null, + }, + }); +} + /** * Persist a progress snapshot onto the mirror `ImportJob` row. The * worker calls this every `PROGRESS_TICK_RECORDS` records parsed + @@ -151,6 +209,7 @@ export async function handleAppleHealthImport( pgBossJobId: job.id, status: "queued", uploadBytes, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); } @@ -348,6 +407,8 @@ const LIVE_PG_BOSS_STATES = new Set(["active", "created", "retry"]); * it last shut down — flip it to `failed` with `interrupted_by_restart` * so the operator can re-upload (and, post-issue-#486, so the kick-off * dedup no longer short-circuits future re-uploads onto a dead job). + * Revision-1 rows are deliberately excluded: during a rolling deployment + * their backing jobs live on the legacy queue and remain owned by old workers. * * The reconcile is deliberately NOT unconditional. In a multi-replica * or rolling-deploy topology a booting worker must not flip a row that @@ -373,7 +434,10 @@ const LIVE_PG_BOSS_STATES = new Set(["active", "created", "retry"]); export async function reconcileOrphanImportJobs(): Promise { const prisma = getWorkerPrisma(); const candidates = await prisma.importJob.findMany({ - where: { status: { in: ["unpacking", "parsing", "upserting"] } }, + where: { + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + status: { in: ["unpacking", "parsing", "upserting"] }, + }, select: { id: true, pgBossJobId: true, updatedAt: true }, }); if (candidates.length === 0) return; @@ -395,7 +459,7 @@ export async function reconcileOrphanImportJobs(): Promise { let live = false; try { const job = await boss.getJobById( - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, row.pgBossJobId, ); live = job !== null && LIVE_PG_BOSS_STATES.has(job.state); diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 6575dec33..2a522fee4 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -38,9 +38,11 @@ import { type IntakeAutoSkipPayload, } from "@/lib/jobs/intake-auto-skip"; import { - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, + APPLE_HEALTH_IMPORT_LEGACY_QUEUE, APPLE_HEALTH_IMPORT_CONCURRENCY, handleAppleHealthImport, + migrateLegacyAppleHealthImport, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; import { @@ -304,7 +306,8 @@ const allQueues = [ // this entry the schedule silently no-ops and pending rows older than // 24 h pile up unflipped. INTAKE_AUTO_SKIP_QUEUE, - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, + APPLE_HEALTH_IMPORT_LEGACY_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 @@ -472,9 +475,10 @@ const schedules: ScheduleEntry[] = [ * become one. That is exactly the class of silent work-dropping a policy is * supposed to prevent, so the queue keeps `standard` until the enqueue side * gives the explicit-range variant a key of its own. - * - APPLE_HEALTH_IMPORT_QUEUE, DATA_BACKUP_QUEUE and PR_DETECTION_QUEUE send - * keylessly by design (each import / backup / detection run is a distinct - * unit of work). A policy would coalesce independent runs. + * - APPLE_HEALTH_IMPORT_V2_QUEUE, APPLE_HEALTH_IMPORT_LEGACY_QUEUE, + * DATA_BACKUP_QUEUE and PR_DETECTION_QUEUE send keylessly by design (each + * import / backup / detection run is a distinct unit of work). A policy + * would coalesce independent runs. */ const queuePolicies: QueuePolicyTable = { // Per-user, boot- and cron-discovery driven, self-converging one-shots. The @@ -672,7 +676,7 @@ export async function registerMaintenanceQueues( // caps at 1 because the parse loop is CPU-bound and a concurrent // second 1 GB import would race the first for RSS. await boss.work( - APPLE_HEALTH_IMPORT_QUEUE, + APPLE_HEALTH_IMPORT_V2_QUEUE, { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY }, async (jobs) => { // pg-boss v12 work callbacks always receive an array (batched @@ -683,6 +687,18 @@ export async function registerMaintenanceQueues( } }, ); + // Drain pre-revision-2 backlog by moving each claimed legacy job onto the + // isolated v2 queue. This preserves the ImportJob status id without letting + // the current parser execute under a revision-1 mirror. + await boss.work( + APPLE_HEALTH_IMPORT_LEGACY_QUEUE, + { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + await migrateLegacyAppleHealthImport(job); + } + }, + ); // v1.8.2 — duplicate dose-slot cleanup worker. The boot enqueue helper // below sends one job per user holding two live intake rows that snap diff --git a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts index e27f1cde3..0d1ba4e8b 100644 --- a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts +++ b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts @@ -356,7 +356,13 @@ describe("drainPerSampleCumulative — cutoffHours", () => { // externalId first, then adopts it in place (`update`) or mints a fresh row // (`create`). describe("drainPerSampleCumulative — late-sync fold into existing total", () => { - function buildFoldMock(existingTotal: number | null) { + function buildFoldMock( + existingTotal: number | null, + aggregationProvenance: + | "HEALTHKIT_STATISTICS" + | "EXPORT_XML_SOURCE_MAX" + | "LEGACY_UNKNOWN" = "LEGACY_UNKNOWN", + ) { const findManyUser = vi .fn() .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]); @@ -414,7 +420,11 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = if ("externalId" in args.where) { return existingTotal === null ? null - : { id: "stats-row", value: existingTotal }; + : { + id: "stats-row", + value: existingTotal, + aggregationProvenance, + }; } if ("measuredAt" in args.where) return null; // no index-B collision return { unit: "count" }; // resolveCanonicalUnit @@ -436,6 +446,7 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = } as unknown as PrismaClient, update, create, + deleteMany, }; } @@ -451,6 +462,10 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = expect(update).toHaveBeenCalledTimes(1); const updateArg = update.mock.calls[0]?.[0] as { data: { value: number } }; expect(updateArg.data.value).toBe(9500); + expect(updateArg.data).toMatchObject({ + aggregationProvenance: "LEGACY_UNKNOWN", + syncVersion: { increment: 1 }, + }); }); it("mints the partial sum when no collapsed total exists yet", async () => { @@ -464,8 +479,29 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = expect(create).toHaveBeenCalledTimes(1); const createArg = create.mock.calls[0]?.[0] as { data: { value: number } }; expect(createArg.data.value).toBe(500); + expect(createArg.data).toMatchObject({ + aggregationProvenance: "LEGACY_UNKNOWN", + }); }); + it.each(["HEALTHKIT_STATISTICS", "EXPORT_XML_SOURCE_MAX"] as const)( + "drains late source rows without changing %s authority", + async (aggregationProvenance) => { + const { prisma, update, create, deleteMany } = buildFoldMock( + 9_000, + aggregationProvenance, + ); + + await drainPerSampleCumulative(prisma, { log: () => {} }); + + expect(update).not.toHaveBeenCalled(); + expect(create).not.toHaveBeenCalled(); + expect(deleteMany).toHaveBeenCalledWith({ + where: { id: { in: ["late-1", "late-2"], not: "stats-row" } }, + }); + }, + ); + it("recomputes the DAY rollup bucket for the drained day so the stale sum self-heals", async () => { vi.mocked(recomputeBucketsForMeasurement).mockClear(); const { prisma } = buildFoldMock(null); @@ -745,11 +781,10 @@ describe("drainPerSampleCumulative — concurrent fold does not double-count (F2 // The advisory lock fired for the bucket. expect($queryRaw).toHaveBeenCalledTimes(1); - // The row is adopted in place and left at 9000 — NOT 9500. Pre-fix the - // merge added the stale `reducedValue` (500) onto the committed total. + // The row is left at 9000 — NOT 9500 — without a non-material update or + // syncVersion bump. Pre-fix the merge added the stale reducedValue (500) + // onto the committed total. expect(create).not.toHaveBeenCalled(); - expect(update).toHaveBeenCalledTimes(1); - const updateArg = update.mock.calls[0]?.[0] as { data: { value: number } }; - expect(updateArg.data.value).toBe(9000); + expect(update).not.toHaveBeenCalled(); }); }); diff --git a/src/lib/measurements/__tests__/import-apple-health-export.test.ts b/src/lib/measurements/__tests__/import-apple-health-export.test.ts index 47ca718b4..389ba7d41 100644 --- a/src/lib/measurements/__tests__/import-apple-health-export.test.ts +++ b/src/lib/measurements/__tests__/import-apple-health-export.test.ts @@ -15,7 +15,7 @@ vi.mock("@/lib/arrivals/emit-shared", () => ({ emitDataArrival: vi.fn().mockResolvedValue(undefined), })); -import { Prisma } from "@/generated/prisma/client"; +import { Prisma, type PrismaClient } from "@/generated/prisma/client"; import { hashSampleKey, parseRecordValue, @@ -27,6 +27,10 @@ import { import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; import { emitDataArrival } from "@/lib/arrivals/emit-shared"; +import { + APPLE_HEALTH_TYPE_MAP, + CUMULATIVE_HK_TYPES, +} from "../apple-health-mapping"; beforeEach(() => { vi.clearAllMocks(); @@ -173,10 +177,42 @@ function makeFakePrisma( workouts.push({ id: `w${nextId++}`, ...create }); }; - return { + const fake = { _measurements: measurements, _workouts: workouts, + $queryRaw: async () => [], + $executeRawUnsafe: async () => 0, + $transaction: async ( + callback: (tx: unknown) => Promise, + ): Promise => callback(fake), measurement: { + create: async ({ data }: { data: Row }) => { + injectMeasurementRace(data); + if ( + measurements.some( + (measurement) => + (measurement.userId === data.userId && + measurement.type === data.type && + measurement.source === data.source && + measurement.externalId === data.externalId) || + naturalKeyMatch(measurement, data), + ) + ) { + throw new Prisma.PrismaClientKnownRequestError( + "Unique constraint failed", + { code: "P2002", clientVersion: "test" }, + ); + } + const created: Row = { + id: `m${nextId++}`, + syncVersion: 1, + deletedAt: null, + sleepStage: null, + ...data, + }; + measurements.push(created); + return created; + }, createManyAndReturn: async ({ data }: { data: Row[] }) => { const returned: Row[] = []; for (const create of data) { @@ -210,20 +246,61 @@ function makeFakePrisma( }: { where: { userId: string; - source: string; - OR: Array>; + source?: unknown; + OR?: Array>; }; }) => { - return measurements - .filter((m) => { - if (m.userId !== where.userId || m.source !== where.source) - return false; - return where.OR.some( - (clause) => - m.type === clause.type && m.externalId === clause.externalId, - ); - }) - .map((m) => ({ type: m.type, externalId: m.externalId })); + const externalClause = where.OR?.find( + (clause) => typeof clause.externalId === "string", + ); + const externalId = + typeof externalClause?.externalId === "string" + ? externalClause.externalId + : null; + if ( + externalId?.startsWith("stats:") && + measurementRaceExternalIds.has(externalId) && + !injectedMeasurementRaces.has(externalId) + ) { + const naturalClause = where.OR?.find( + (clause) => clause.measuredAt instanceof Date, + ); + injectedMeasurementRaces.add(externalId); + measurements.push({ + id: `m${nextId++}`, + userId: where.userId, + type: externalClause?.type, + source: externalClause?.source, + externalId, + measuredAt: naturalClause?.measuredAt, + sleepStage: naturalClause?.sleepStage ?? null, + deletedAt: null, + syncVersion: 1, + aggregationProvenance: "HEALTHKIT_STATISTICS", + }); + } + return measurements.filter((measurement) => { + if (measurement.userId !== where.userId) return false; + if ( + where.source !== undefined && + measurement.source !== where.source + ) { + return false; + } + if (!where.OR) return true; + return where.OR.some( + (clause) => + (clause.type === undefined || measurement.type === clause.type) && + (clause.source === undefined || + measurement.source === clause.source) && + (clause.externalId === undefined || + measurement.externalId === clause.externalId) && + (clause.measuredAt === undefined || + sameMeasuredAt(measurement.measuredAt, clause.measuredAt)) && + (clause.sleepStage === undefined || + sameStage(measurement.sleepStage, clause.sleepStage)), + ); + }); }, findUnique: async ({ where, @@ -244,7 +321,15 @@ function makeFakePrisma( // Natural-key probe used by the rescue path. Tombstone-inclusive: no // `deletedAt` filter is applied here or by the caller. findFirst: async ({ where }: { where: Record }) => { - const match = measurements.find((m) => naturalKeyMatch(m, where)); + const excludedId = + typeof where.id === "object" && where.id !== null && "not" in where.id + ? where.id.not + : undefined; + const match = measurements.find( + (measurement) => + measurement.id !== excludedId && + naturalKeyMatch(measurement, where), + ); return match ? { id: match.id } : null; }, update: async ({ @@ -276,7 +361,19 @@ function makeFakePrisma( { code: "P2025", clientVersion: "test" }, ); } - measurements[idx] = { ...measurements[idx], ...data }; + const syncVersion = + typeof data.syncVersion === "object" && + data.syncVersion !== null && + "increment" in data.syncVersion && + typeof data.syncVersion.increment === "number" + ? Number(measurements[idx].syncVersion ?? 1) + + data.syncVersion.increment + : data.syncVersion; + measurements[idx] = { + ...measurements[idx], + ...data, + ...(syncVersion === undefined ? {} : { syncVersion }), + }; return measurements[idx]; }, upsert: async ({ @@ -398,6 +495,7 @@ function makeFakePrisma( }, }, }; + return fake; } describe("parseRecordValue", () => { @@ -494,6 +592,143 @@ describe("hashSampleKey", () => { expect(a).not.toBe(b); }); }); +const cumulativeMappings = Object.values(APPLE_HEALTH_TYPE_MAP).filter( + (mapping) => CUMULATIVE_HK_TYPES.has(mapping.measurementType), +); + +function cumulativeExportXml( + hkIdentifier: string, + hkUnit: string, + rows: Array<{ value: number; sourceName: string; hour: number }>, +): string { + return ` + +${rows + .map( + ({ value, sourceName, hour }) => ` `, + ) + .join("\n")} +`; +} + +async function importCumulativeFixture(xml: string) { + const tmp = mkdtempSync(join(tmpdir(), "healthlog-parser-cumulative-test-")); + const xmlPath = join(tmp, "export.xml"); + writeFileSync(xmlPath, xml); + const prisma = makeFakePrisma(); + const result = await streamParseExportXml({ + xmlPath, + userId: "user-cumulative", + userTimezone: "Europe/Berlin", + // The fake implements the importer-facing subset of PrismaClient. + prisma: prisma as unknown as PrismaClient, + }); + return { prisma, result }; +} + +describe("streamParseExportXml — cumulative source-day estimates", () => { + it("covers exactly the six cumulative HealthKit measurement types", () => { + expect(CUMULATIVE_HK_TYPES.size).toBe(6); + expect( + new Set(cumulativeMappings.map((mapping) => mapping.measurementType)), + ).toEqual(CUMULATIVE_HK_TYPES); + }); + + it("stores 8,526 rather than source-blind 15,608 for overlapping iPhone and Zepp steps", async () => { + const { prisma, result } = await importCumulativeFixture( + cumulativeExportXml("HKQuantityTypeIdentifierStepCount", "count", [ + { value: 8_526, sourceName: "iPhone", hour: 8 }, + { value: 7_082, sourceName: "Zepp", hour: 9 }, + ]), + ); + + expect(prisma._measurements).toHaveLength(1); + expect(prisma._measurements[0]).toMatchObject({ + value: 8_526, + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: 2, + }); + expect(prisma._measurements[0]?.aggregationSelectedSourceHash).toMatch( + /^[a-f0-9]{64}$/, + ); + expect(result.cumulativeEstimates).toEqual({ days: 1, rows: 1 }); + }); + + it.each(cumulativeMappings)( + "selects the largest source-day subtotal for $measurementType", + async (mapping) => { + const { prisma } = await importCumulativeFixture( + cumulativeExportXml(mapping.hkIdentifier, mapping.hkUnit, [ + { value: 30, sourceName: "Source A", hour: 8 }, + { value: 20, sourceName: "Source B", hour: 9 }, + ]), + ); + + expect(prisma._measurements).toHaveLength(1); + expect(prisma._measurements[0]).toMatchObject({ + type: mapping.measurementType, + value: 30, + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: 2, + }); + }, + ); + + it.each(cumulativeMappings)( + "sums non-overlapping $measurementType rows from the same source", + async (mapping) => { + const { prisma } = await importCumulativeFixture( + cumulativeExportXml(mapping.hkIdentifier, mapping.hkUnit, [ + { value: 30, sourceName: "Source A", hour: 8 }, + { value: 20, sourceName: "Source A", hour: 9 }, + ]), + ); + + expect(prisma._measurements).toHaveLength(1); + expect(prisma._measurements[0]).toMatchObject({ + type: mapping.measurementType, + value: 50, + aggregationContributorCount: 1, + }); + }, + ); + it("folds unattributed cumulative records into one bounded source bucket", async () => { + const { prisma } = await importCumulativeFixture(` + + + +`); + + expect(prisma._measurements).toHaveLength(1); + expect(prisma._measurements[0]).toMatchObject({ + value: 100, + aggregationContributorCount: 1, + }); + }); + + it("rejects a source-day subtotal outside the aggregate range", async () => { + const { prisma, result } = await importCumulativeFixture( + cumulativeExportXml("HKQuantityTypeIdentifierStepCount", "count", [ + { value: 150_000, sourceName: "Source A", hour: 8 }, + { value: 150_000, sourceName: "Source A", hour: 9 }, + ]), + ); + + expect(prisma._measurements).toHaveLength(0); + expect(result.unknown["ACTIVITY_STEPS::aggregate_out_of_range"]).toBe(1); + expect(result.cumulativeEstimates).toEqual({ days: 0, rows: 0 }); + }); +}); describe("streamParseExportXml — end-to-end", () => { it("ingests the tiny fixture into per-type / workout / deferred / unknown buckets", async () => { @@ -615,8 +850,8 @@ describe("streamParseExportXml — statement-level insertion identity", () => { xmlPath, userId: "user-1", userTimezone: "Europe/Berlin", - // eslint-disable-next-line @typescript-eslint/no-explicit-any - prisma: prisma as any, + // The fake implements the importer-facing subset of PrismaClient. + prisma: prisma as unknown as PrismaClient, spotBatchSize: 10, }); @@ -625,9 +860,11 @@ describe("streamParseExportXml — statement-level insertion identity", () => { inserted: 1, updated: 0, }); + // The concurrently visible native statistics row is authoritative, so the + // XML estimate is an intentional no-op. expect(result.perType.ACTIVITY_STEPS).toMatchObject({ inserted: 0, - updated: 1, + updated: 0, }); const arrivalRows = vi.mocked(emitInsertedMeasurementArrivals).mock @@ -907,7 +1144,7 @@ describe("streamParseExportXml — memory ceiling", () => { ` unit="count"` + ` startDate="${ts}"` + ` endDate="${ts}"` + - ` value="${(i % 200) + 1}"/>\n`, + ` value="1"/>\n`, ); } writeFileSync(xmlPath, header + rows.join("") + footer); @@ -931,6 +1168,11 @@ describe("streamParseExportXml — memory ceiling", () => { // The cumulative fold collapses all 10 000 records into one daily // bucket (they all land on the same date in the fixture). expect(result.perType.ACTIVITY_STEPS?.inserted).toBe(1); + expect(prisma._measurements).toHaveLength(1); + expect(prisma._measurements[0]).toMatchObject({ + value: 10_000, + aggregationContributorCount: 1, + }); // Heap delta should be modest — we tolerate a generous 100 MB so a // GC-related spike doesn't flake the test, but a regression that diff --git a/src/lib/measurements/__tests__/reconcile-external-measurement.test.ts b/src/lib/measurements/__tests__/reconcile-external-measurement.test.ts index 4034767ec..1a755dc72 100644 --- a/src/lib/measurements/__tests__/reconcile-external-measurement.test.ts +++ b/src/lib/measurements/__tests__/reconcile-external-measurement.test.ts @@ -39,6 +39,9 @@ function row(overrides: Record = {}): Record { notesEncrypted: null, externalId: "sleep:new", externalSourceVersion: null, + aggregationProvenance: null, + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, glucoseContext: null, sleepStage: "DEEP", rhythmClassification: null, @@ -210,6 +213,167 @@ describe("reconcileExternalMeasurement", () => { expect(measurement.update).not.toHaveBeenCalled(); }); + it("keeps a higher-authority native aggregate unchanged when XML arrives later", async () => { + const native = row({ + id: "native-stats", + type: "ACTIVITY_STEPS", + value: 8_600, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T10:00:00.000Z"), + sleepStage: null, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + }); + const { tx, measurement } = transaction([native]); + + const verdict = await reconcileExternalMeasurement( + tx, + desired({ + type: "ACTIVITY_STEPS", + value: 8_526, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T10:00:00.000Z"), + sleepStage: null, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: 2, + aggregationSelectedSourceHash: "source-hash", + }), + { exactExternalMatch: "update" }, + ); + + expect(verdict).toMatchObject({ + status: "duplicate", + row: { id: "native-stats" }, + }); + expect(measurement.update).not.toHaveBeenCalled(); + }); + + it("retires a lower-authority external collision without changing the native natural-key row", async () => { + const xmlExternal = row({ + id: "xml-external", + type: "ACTIVITY_STEPS", + value: 8_526, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T09:00:00.000Z"), + sleepStage: null, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + }); + const nativeNatural = row({ + id: "native-natural", + type: "ACTIVITY_STEPS", + value: 8_600, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T10:00:00.000Z"), + sleepStage: null, + externalId: "stats:legacy-native-id", + externalSourceVersion: "native-v1", + deviceType: "Apple Watch", + aggregationProvenance: "HEALTHKIT_STATISTICS", + }); + const { tx, measurement } = transaction([xmlExternal, nativeNatural]); + + const verdict = await reconcileExternalMeasurement( + tx, + desired({ + type: "ACTIVITY_STEPS", + value: 8_526, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T10:00:00.000Z"), + sleepStage: null, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + }), + { exactExternalMatch: "update" }, + ); + + expect(verdict).toMatchObject({ + status: "duplicate", + row: { + id: "native-natural", + aggregationProvenance: "HEALTHKIT_STATISTICS", + }, + retiredCollisionId: "xml-external", + }); + expect(measurement.update).toHaveBeenCalledTimes(1); + expect(measurement.update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: "xml-external" }, + data: expect.objectContaining({ + measuredAt: new Date(0), + externalId: + "retired:xml-external:stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + deletedAt: expect.any(Date), + syncVersion: { increment: 1 }, + }), + }), + ); + }); + + it("allows native statistics to repair a legacy aggregate", async () => { + const legacy = row({ + id: "legacy-stats", + type: "ACTIVITY_STEPS", + value: 15_608, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T10:00:00.000Z"), + sleepStage: null, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + aggregationProvenance: "LEGACY_UNKNOWN", + }); + const { tx, measurement } = transaction([legacy]); + measurement.update.mockResolvedValue( + row({ + ...legacy, + value: 8_600, + aggregationProvenance: "HEALTHKIT_STATISTICS", + }), + ); + + const verdict = await reconcileExternalMeasurement( + tx, + desired({ + type: "ACTIVITY_STEPS", + value: 8_600, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T10:00:00.000Z"), + sleepStage: null, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + }), + { exactExternalMatch: "update" }, + ); + + expect(verdict).toMatchObject({ + status: "updated", + row: { id: "legacy-stats" }, + }); + expect(measurement.update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: "legacy-stats" }, + data: expect.objectContaining({ + value: 8_600, + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + syncVersion: { increment: 1 }, + }), + }), + ); + }); + it("rolls back its savepoint and returns failed for a non-benign write error", async () => { const existing = row({ id: "exact-row", diff --git a/src/lib/measurements/drain-per-sample-cumulative.ts b/src/lib/measurements/drain-per-sample-cumulative.ts index bf9965844..9fb503f77 100644 --- a/src/lib/measurements/drain-per-sample-cumulative.ts +++ b/src/lib/measurements/drain-per-sample-cumulative.ts @@ -6,20 +6,18 @@ * * Scope per `CUMULATIVE_HK_TYPES`: * ACTIVITY_STEPS, ACTIVE_ENERGY_BURNED, FLIGHTS_CLIMBED, - * WALKING_RUNNING_DISTANCE, TIME_IN_DAYLIGHT + * WALKING_RUNNING_DISTANCE, TIME_IN_DAYLIGHT, FALL_COUNT * * Per user × type × calendar day (anchored to `User.timezone`): - * 1. SELECT all `Measurement` rows with `source = 'APPLE_HEALTH'` and - * `type = ` and `measuredAt` within that user's - * calendar day boundary and `externalId NOT LIKE 'stats:%'`. - * 2. If 0 rows → continue. - * 3. If 1 row whose externalId already follows the `stats:...` - * shape → continue (already collapsed). - * 4. SUM the values; pick canonical timestamp = midday UTC of the - * user's calendar day (matches the Withings activity-sync - * convention per R-A §5 / W17b). - * 5. UPSERT a row with `externalId = dailyStatsExternalId(...)`, - * `value = sumValue`, `measuredAt = canonicalTimestamp`. + * 1. SELECT live APPLE_HEALTH rows for the cumulative type and day. + * 2. Bucket only non-`stats:` source rows and sum their values. + * 3. Resolve the canonical `stats::` row under an advisory + * transaction lock. + * 4. If it is HEALTHKIT_STATISTICS or EXPORT_XML_SOURCE_MAX, leave its + * value/provenance untouched and drain only the source rows. + * 5. Otherwise preserve legacy late-increment semantics, explicitly stamp + * LEGACY_UNKNOWN, and increment syncVersion; fresh/adopted rows are also + * LEGACY_UNKNOWN. * 6. DELETE the original per-sample rows in the same transaction. * * Designed to be invoked by both: @@ -32,7 +30,11 @@ * committing. */ import { Prisma } from "@/generated/prisma/client"; -import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import type { + MeasurementAggregationProvenance, + MeasurementType, + PrismaClient, +} from "@/generated/prisma/client"; import { isP2002 as isUniqueConstraintViolation } from "@/lib/prisma-errors"; import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; @@ -71,6 +73,11 @@ export type { PerSampleRow }; /** Prefix marking an already-collapsed daily-stats row. */ const DAILY_STATS_PREFIX = "stats:"; +const AUTHORITATIVE_AGGREGATION_PROVENANCE = + new Set([ + "HEALTHKIT_STATISTICS", + "EXPORT_XML_SOURCE_MAX", + ]); /** * v1.4.38 — canonical cutoff for the nightly scheduled drain. Rows @@ -284,7 +291,7 @@ export async function drainPerSampleCumulative( // present, this is the existing collapsed daily total. const eidRow = await tx.measurement.findFirst({ where: { userId, type, source: "APPLE_HEALTH", externalId }, - select: { id: true, value: true }, + select: { id: true, value: true, aggregationProvenance: true }, orderBy: { id: "asc" }, }); // Index-B row: occupies the canonical local-noon instant @@ -297,7 +304,7 @@ export async function drainPerSampleCumulative( measuredAt: canonicalTimestamp, sleepStage: null, }, - select: { id: true }, + select: { id: true, aggregationProvenance: true }, orderBy: { id: "asc" }, }); @@ -305,66 +312,63 @@ export async function drainPerSampleCumulative( // and stamping the target externalId onto it would collide with it. const adoptTarget = eidRow ?? slotRow; - // Fold semantics. When the index-A `stats:` total already exists, - // late per-sample rows are GENUINE additional readings, so the - // correct merge is `existing + late sum` (a blind overwrite with the - // partial late sum would shrink the day — the original samples are - // already hard-deleted). When adopting a NON-stats slot row (a - // per-sample row that fell on local noon, itself part of this - // bucket), its value is already inside `reducedValue`, so the merged - // value is just `reducedValue`. A fresh day mints `reducedValue`. - // - // Concurrency guard (paired with the advisory lock above): a sibling - // fold that committed just before this one may have already merged - // some or all of these per-sample rows into `eidRow` and hard-deleted - // them. Trusting the pre-scan `reducedValue` for the increment would - // then re-add a sibling's already-folded work → the permanent 2×. - // Re-read the still-present contributors inside the locked - // transaction: whatever survives is the genuine un-folded increment - // (all of `reducedValue` on a clean single run; zero for a - // serialised-second run whose rows the winner already drained). + // Only legacy aggregates may retain the historical late-increment + // behaviour. Native HealthKit statistics and export.xml source-max + // estimates are authoritative results from their own pipelines: + // this maintenance drain may remove their legacy source rows, but + // must never alter their value or provenance. + const adoptProvenance = adoptTarget?.aggregationProvenance; + const protectedAggregate = + adoptProvenance != null && + AUTHORITATIVE_AGGREGATION_PROVENANCE.has(adoptProvenance); + + // Re-read contributors under the advisory lock. A serialized loser + // sees zero live rows and therefore performs no canonical update. const liveIncrement = - eidRow !== null + eidRow !== null && !protectedAggregate ? ( await tx.measurement.findMany({ where: { id: { in: sourceRowIds } }, select: { value: true }, }) - ).reduce((sum, r) => sum + r.value, 0) + ).reduce((sum, row) => sum + row.value, 0) : 0; const mergedValue = eidRow !== null ? eidRow.value + liveIncrement : reducedValue; let canonicalRowId: string; if (adoptTarget) { - // Pin the adopted row to local-noon only when the canonical slot - // is free or already this row — a DIFFERENT row occupying the slot - // would otherwise collide on index B. In that rare case (a tz/DST - // shift between mints) the row keeps its existing instant; the - // `stats:` externalId is the identity, measuredAt is secondary. - const slotIsFreeForTarget = - slotRow === null || slotRow.id === adoptTarget.id; - await tx.measurement.update({ - where: { id: adoptTarget.id }, - data: { - value: mergedValue, - externalId, - deletedAt: null, - ...(slotIsFreeForTarget - ? { measuredAt: canonicalTimestamp } - : {}), - }, - }); canonicalRowId = adoptTarget.id; + if (!protectedAggregate && (eidRow === null || liveIncrement > 0)) { + // Pin the adopted row to local-noon only when the canonical slot + // is free or already this row. A different row occupying the + // slot would collide on the natural identity. + const slotIsFreeForTarget = + slotRow === null || slotRow.id === adoptTarget.id; + await tx.measurement.update({ + where: { id: adoptTarget.id }, + data: { + value: mergedValue, + externalId, + deletedAt: null, + aggregationProvenance: "LEGACY_UNKNOWN", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + syncVersion: { increment: 1 }, + ...(slotIsFreeForTarget + ? { measuredAt: canonicalTimestamp } + : {}), + }, + }); + } } else { const created = await tx.measurement.create({ data: { userId, type, value: reducedValue, - // pick the canonical unit from an existing row; the per-sample - // rows all carry the same unit on a given type. dayRows always - // has ≥1 row (empty buckets are skipped). + // Pick the canonical unit from an existing row; the per-sample + // rows all carry the same unit on a given type. unit: dayRows[0]?.value !== undefined ? await resolveCanonicalUnit(tx, userId, type) @@ -372,6 +376,7 @@ export async function drainPerSampleCumulative( source: "APPLE_HEALTH", measuredAt: canonicalTimestamp, externalId, + aggregationProvenance: "LEGACY_UNKNOWN", }, select: { id: true }, }); diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index 6c607a168..030cd5b10 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -4,18 +4,18 @@ * Reads an `export.xml` byte-stream via `sax` (event-driven SAX, no * DOM) and folds every ``, ``, and `` * element into the row shape the existing `Measurement` and - * `Workout` models expect. Cumulative-quantity `` rows - * (steps, energy, distance, flights, daylight) collapse into one - * `stats::` `Measurement` per user-local day - * — mirroring iOS's `HKStatisticsCollectionQuery` daily-aggregation - * convention locked in v1.4.30. Spot rows (BP, weight, HRV, …) - * survive verbatim, keyed by `HKSample.uuid` when present. + * `Workout` models expect. Cumulative-quantity `` rows are folded by + * `(type, local day, hashed source identity)`, summed within each source-day, + * then reduced to the largest source subtotal. The resulting + * `stats::` row is explicitly an export estimate; native + * HealthKit statistics remain authoritative. * - * Memory profile: SAX callbacks fire as the byte cursor advances, - * so peak RSS stays bounded regardless of the input file size. The - * cumulative-bucket map holds at most one `(type, day)` entry per - * observed day per cumulative type — a 10-year export with five - * cumulative types lands at ~18 000 entries (~1 MB). + * Spot rows (BP, weight, HRV, …) survive verbatim, keyed by `HKSample.uuid` + * when present. SAX callbacks fire as the byte cursor advances, so peak RSS + * stays bounded regardless of input size. The cumulative map grows with the + * observed `(type, day, source hash)` combinations; every record without source + * metadata shares one bounded bucket, and raw source/device labels are never + * retained. * * Locks per `.planning/research/v1434-r-1-xml-import.md` §6. */ @@ -39,6 +39,7 @@ import { dayKeyForUserTz, canonicalDailyTimestamp, } from "@/lib/measurements/drain-per-sample-cumulative"; +import { reconcileExternalMeasurement } from "@/lib/measurements/reconcile-external-measurement"; import { resolveHkWorkoutSportType } from "@/lib/measurements/hk-workout-activity-type-map"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"; @@ -113,6 +114,12 @@ export interface ImportJobResult { cycle: CycleImportStats; deferred: Record; unknown: Record; + cumulativeEstimates: { + /** Distinct local calendar days containing at least one estimated total. */ + days: number; + /** Estimated `(measurement type, local day)` aggregate rows considered. */ + rows: number; + }; totals: { recordsRead: number; rowsUpserted: number; @@ -147,6 +154,8 @@ interface PreparedWorkout { metadata: Prisma.JsonValue | null; } +type CumulativeSourceSubtotals = Map; + /** Per-type running stats accumulator. */ interface MutableTypeStat { read: number; @@ -182,6 +191,34 @@ export function hashSampleKey( ); } +const UNATTRIBUTED_CUMULATIVE_SOURCE_HASH = createHash("sha256") + .update(JSON.stringify(["unattributed"])) + .digest("hex"); + +/** + * Hash the source tuple used only to keep overlapping export.xml contributors + * separate. Records without source metadata share one stable bucket so parser + * state remains bounded by actual source cardinality rather than record count. + * Raw source/device labels never leave the parser. + */ +export function hashCumulativeSourceIdentity( + sourceName: string | undefined, + sourceVersion: string | undefined, + device: string | undefined, +): string { + const sourceTuple = [ + sourceName?.trim() ?? "", + sourceVersion?.trim() ?? "", + device?.trim() ?? "", + ]; + if (!sourceTuple.some(Boolean)) { + return UNATTRIBUTED_CUMULATIVE_SOURCE_HASH; + } + return createHash("sha256") + .update(JSON.stringify(["source", ...sourceTuple])) + .digest("hex"); +} + /** * Parse the Apple `Record` `value` attribute. Quantity records carry * a numeric string; sleep-analysis records carry the symbolic name. @@ -235,7 +272,7 @@ export interface StreamParseInput { /** IANA timezone, used to anchor cumulative-type day-keys. */ userTimezone: string; /** Prisma client to flush rows through. */ - prisma: Pick; + prisma: Pick; /** * Live progress hook. Called every `PROGRESS_TICK_RECORDS` records * read, and once on terminal `done`. Best-effort; the parser @@ -305,8 +342,11 @@ export async function streamParseExportXml( protectionUsed?: boolean; } | null = null; - // Cumulative-type fold: type -> dayKey -> running sum. - const cumulativeBucket = new Map>(); + // Cumulative fold: type -> local day -> hashed source identity -> subtotal. + const cumulativeBucket = new Map< + MeasurementType, + Map + >(); // Spot-row batch awaiting flush. const spotBatch: PreparedMeasurement[] = []; // Workout-row batch awaiting flush. @@ -314,6 +354,8 @@ export async function streamParseExportXml( let recordsRead = 0; let rowsUpserted = 0; + const cumulativeEstimatedDays = new Set(); + let cumulativeEstimatedRows = 0; // Per R-1 §8 the percent stays best-effort and may remain null // until the parser sees the closing `` tag. We don't // currently mutate this in v1.4.34 — the iOS app keeps polling on @@ -617,56 +659,74 @@ export async function streamParseExportXml( const flushCumulativeBuckets = async (): Promise => { for (const [type, byDay] of cumulativeBucket.entries()) { - // Re-resolve the HK identifier from the mapping table — the - // bucket keys are MeasurementType, but the externalId carries - // the HK identifier so a future re-import collides on the - // same string. + // Re-resolve the HK identifier from the mapping table — the bucket keys + // are MeasurementType, but the shared externalId carries the HK + // identifier used by native HealthKit statistics. const mapping = Object.values(APPLE_HEALTH_TYPE_MAP).find( - (m) => m.measurementType === type, + (candidate) => candidate.measurementType === type, ); if (!mapping) continue; const stat = bumpStat(type); - for (const [dayKey, sum] of byDay.entries()) { + for (const [dayKey, bySource] of byDay.entries()) { + let selected: [sourceHash: string, subtotal: number] | undefined; + for (const entry of bySource.entries()) { + if ( + !selected || + entry[1] > selected[1] || + (entry[1] === selected[1] && entry[0] < selected[0]) + ) { + selected = entry; + } + } + if (!selected) continue; + const [selectedSourceHash, selectedSubtotal] = selected; + if (validateMeasurementRange(type, selectedSubtotal) !== null) { + unknown[`${type}::aggregate_out_of_range`] = + (unknown[`${type}::aggregate_out_of_range`] ?? 0) + 1; + continue; + } const externalId = dailyStatsExternalId(mapping.hkIdentifier, dayKey); const measuredAt = canonicalDailyTimestamp(dayKey, userTimezone); const rowStart = Date.now(); - const created = await prisma.measurement.createManyAndReturn({ - data: [ + const verdict = await prisma.$transaction((tx) => + reconcileExternalMeasurement( + tx, { userId, type, - value: sum, + value: selectedSubtotal, unit: mapping.dbUnit, source: "APPLE_HEALTH", measuredAt, externalId, + externalSourceVersion: null, + sleepStage: null, + deviceType: null, + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: bySource.size, + aggregationSelectedSourceHash: selectedSourceHash, }, - ], - skipDuplicates: true, - select: { id: true }, - }); - if (created.length > 0) { + { exactExternalMatch: "update" }, + ), + ); + + cumulativeEstimatedDays.add(dayKey); + cumulativeEstimatedRows += 1; + if (verdict.status === "inserted") { stat.inserted += 1; - } else { - await prisma.measurement.update({ - where: { - userId_type_source_externalId: { - userId, - type, - source: "APPLE_HEALTH", - externalId, - }, - }, - data: { - value: sum, - measuredAt, - }, - }); + rowsUpserted += 1; + } else if ( + verdict.status === "updated" || + verdict.status === "resurrected" + ) { stat.updated += 1; + rowsUpserted += 1; + } else if (verdict.status === "failed") { + unknown[`${type}::upsert_failed`] = + (unknown[`${type}::upsert_failed`] ?? 0) + 1; } stat.durationMs += Date.now() - rowStart; - rowsUpserted += 1; } } }; @@ -758,12 +818,25 @@ export async function streamParseExportXml( if (CUMULATIVE_HK_TYPES.has(mapped.type)) { const dayKey = dayKeyForUserTz(mapped.takenAt, userTimezone); + const sourceHash = hashCumulativeSourceIdentity( + attrs.sourceName, + attrs.sourceVersion, + attrs.device, + ); let byDay = cumulativeBucket.get(mapped.type); if (!byDay) { byDay = new Map(); cumulativeBucket.set(mapped.type, byDay); } - byDay.set(dayKey, (byDay.get(dayKey) ?? 0) + mapped.value); + let bySource = byDay.get(dayKey); + if (!bySource) { + bySource = new Map(); + byDay.set(dayKey, bySource); + } + bySource.set( + sourceHash, + (bySource.get(sourceHash) ?? 0) + mapped.value, + ); } else { // Spot row: derive a stable externalId, queue for flush. const externalId = hashSampleKey( @@ -1008,6 +1081,10 @@ export async function streamParseExportXml( cycle, deferred, unknown, + cumulativeEstimates: { + days: cumulativeEstimatedDays.size, + rows: cumulativeEstimatedRows, + }, totals: { recordsRead, rowsUpserted, diff --git a/src/lib/measurements/reconcile-external-measurement.ts b/src/lib/measurements/reconcile-external-measurement.ts index d45d998c9..031dda441 100644 --- a/src/lib/measurements/reconcile-external-measurement.ts +++ b/src/lib/measurements/reconcile-external-measurement.ts @@ -1,6 +1,7 @@ import { Prisma, type Measurement, + type MeasurementAggregationProvenance, type Prisma as PrismaTypes, } from "@/generated/prisma/client"; @@ -15,6 +16,7 @@ const rowSelect = { sleepStage: true, externalId: true, deletedAt: true, + aggregationProvenance: true, } satisfies PrismaTypes.MeasurementSelect; export type ExternalMeasurementWrite = Omit< @@ -74,6 +76,22 @@ export interface MeasurementReconciliationOptions { exactExternalMatch?: "update" | "duplicate"; } +const AGGREGATION_AUTHORITY: Readonly< + Record +> = { + LEGACY_UNKNOWN: 1, + EXPORT_XML_SOURCE_MAX: 2, + HEALTHKIT_STATISTICS: 3, +}; + +function aggregationAuthority( + value: MeasurementAggregationProvenance | null | undefined, +): number { + return value === null || value === undefined + ? 0 + : AGGREGATION_AUTHORITY[value]; +} + function errorVerdict(err: unknown): MeasurementReconciliationVerdict { const message = err instanceof Error ? err.message : String(err); if (err instanceof Prisma.PrismaClientKnownRequestError) { @@ -162,6 +180,7 @@ async function lockAndReloadCandidates( async function retireCollision( tx: PrismaTypes.TransactionClient, row: PrismaTypes.MeasurementGetPayload<{ select: typeof rowSelect }>, + releaseExternalId = false, ): Promise { let measuredAt = new Date(0); while ( @@ -183,6 +202,9 @@ async function retireCollision( where: { id: row.id }, data: { measuredAt, + ...(releaseExternalId + ? { externalId: `retired:${row.id}:${row.externalId}` } + : {}), deletedAt: new Date(), syncVersion: { increment: 1 }, }, @@ -240,12 +262,38 @@ export async function reconcileExternalMeasurement( return { status: "inserted", row: created }; } - const canonical = externalHit ?? naturalHit!; + let canonical = externalHit ?? naturalHit!; + if ( + desired.aggregationProvenance != null && + externalHit && + naturalHit && + aggregationAuthority(naturalHit.aggregationProvenance) > + aggregationAuthority(externalHit.aggregationProvenance) + ) { + canonical = naturalHit; + } const redundant = externalHit && naturalHit && externalHit.id !== naturalHit.id - ? naturalHit + ? canonical.id === externalHit.id + ? naturalHit + : externalHit : undefined; const exactExternalMatch = externalHit?.id === naturalHit?.id; + if ( + desired.aggregationProvenance != null && + aggregationAuthority(desired.aggregationProvenance) < + aggregationAuthority(canonical.aggregationProvenance) + ) { + if (redundant) { + await retireCollision(tx, redundant, true); + } + await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`); + return { + status: "duplicate", + row: canonical, + ...(redundant ? { retiredCollisionId: redundant.id } : {}), + }; + } if ( options.exactExternalMatch === "duplicate" && diff --git a/src/lib/medications/scheduling/__tests__/next-due.test.ts b/src/lib/medications/scheduling/__tests__/next-due.test.ts index 2a581b003..99f7c7445 100644 --- a/src/lib/medications/scheduling/__tests__/next-due.test.ts +++ b/src/lib/medications/scheduling/__tests__/next-due.test.ts @@ -344,6 +344,49 @@ describe("computeDisplayDue — open overdue slot", () => { }); }); +describe("computeDisplayDue — canonical availability start", () => { + const weekly = makeDailySchedule({ + id: "sched-weekly", + windowStart: "08:00", + windowEnd: "08:00", + timesOfDay: ["08:00"], + daysOfWeek: "3", + rrule: "FREQ=WEEKLY;BYDAY=WE", + }); + + it("carries the weekly default band's early start for a future slot", () => { + const due = computeDisplayDue({ + medication: makeMedication(), + schedules: [weekly], + now: d("2026-06-09T05:30:00Z"), + userTz: BERLIN, + lastIntakeAt: null, + }); + + expect(due?.at.toISOString()).toBe("2026-06-10T06:00:00.000Z"); + expect(due?.overdue).toBe(false); + expect(due?.availableFrom?.toISOString()).toBe("2026-06-09T05:00:00.000Z"); + }); + + it("uses an explicit weekly dose-window boundary instead of the default day-scale lead", () => { + const due = computeDisplayDue({ + medication: makeMedication(), + schedules: [ + { + ...weekly, + doseWindows: [{ timeOfDay: "08:00", start: "06:00", end: "10:00" }], + }, + ], + now: d("2026-06-09T05:30:00Z"), + userTz: BERLIN, + lastIntakeAt: null, + }); + + expect(due?.at.toISOString()).toBe("2026-06-10T06:00:00.000Z"); + expect(due?.availableFrom?.toISOString()).toBe("2026-06-10T03:00:00.000Z"); + }); +}); + /** * v1.16.9 — an AD-HOC row (`scheduledFor === takenAt`) must never * ±6h-resolve a DIFFERENT slot. The proven failure: twice-daily 08:00 / diff --git a/src/lib/medications/scheduling/next-due.ts b/src/lib/medications/scheduling/next-due.ts index 21cb71e94..c38ae1ffb 100644 --- a/src/lib/medications/scheduling/next-due.ts +++ b/src/lib/medications/scheduling/next-due.ts @@ -146,9 +146,14 @@ export function computeNextDueAt(input: { return earliest; } -/** The instant a medication card should surface, plus its overdue state. */ +/** The instant a medication card should surface, plus its canonical band state. */ export interface DisplayDue { at: Date; + /** + * Earliest instant at which the canonical attribution band accepts this + * slot. This is cadence- and per-dose-window-aware. + */ + availableFrom?: Date; /** * True when `at` is an OPEN overdue slot: its anchor has passed but `now` * is still inside the slot's catch-up band (`anchor < now ≤ overdueEnd`) @@ -190,12 +195,19 @@ export function computeDisplayDue( input: ComputeDisplayDueInput, ): DisplayDue | null { const open = findOpenOverdueSlot(input); - if (open) return { at: open, overdue: true }; + if (open) return { ...open, overdue: true }; const next = computeNextDueAt(input); - return next ? { at: next, overdue: false } : null; + if (!next) return null; + return { + at: next, + availableFrom: findAvailabilityStart(input, next) ?? next, + overdue: false, + }; } -function findOpenOverdueSlot(input: ComputeDisplayDueInput): Date | null { +function findOpenOverdueSlot( + input: ComputeDisplayDueInput, +): { at: Date; availableFrom: Date } | null { const { medication, schedules, now, userTz, lastIntakeAt } = input; if (schedules.length === 0) return null; @@ -208,7 +220,7 @@ function findOpenOverdueSlot(input: ComputeDisplayDueInput): Date | null { if (floor.getTime() >= now.getTime()) return null; const ctx = buildRecurrenceContext({ medication, userTz, lastIntakeAt }); - let latest: Date | null = null; + let latest: { at: Date; availableFrom: Date } | null = null; for (const schedule of schedules) { const { bands } = buildBandsForMedication({ medication, @@ -229,8 +241,57 @@ function findOpenOverdueSlot(input: ComputeDisplayDueInput): Date | null { if (anchor >= now.getTime()) continue; if (now.getTime() > band.overdueEnd.getTime()) continue; if (isResolved(band.at)) continue; - if (latest === null || anchor > latest.getTime()) latest = band.at; + if ( + latest === null || + anchor > latest.at.getTime() || + (anchor === latest.at.getTime() && + band.earlyStart.getTime() < latest.availableFrom.getTime()) + ) { + latest = { at: band.at, availableFrom: band.earlyStart }; + } } } return latest; } + +function findAvailabilityStart( + input: ComputeDisplayDueInput, + at: Date, +): Date | null { + const ctx = buildRecurrenceContext({ + medication: input.medication, + userTz: input.userTz, + lastIntakeAt: input.lastIntakeAt, + }); + const from = new Date(at.getTime() - OVERDUE_LOOKBACK_MS); + let earliest: Date | null = null; + + for (const scheduleRow of input.schedules) { + const schedule = buildCanonicalSchedule(scheduleRow); + const intakeInstants = + schedule.rollingIntervalDays === null + ? input.lastIntakeAt + ? [input.lastIntakeAt] + : [] + : [input.lastIntakeAt, at].filter( + (instant): instant is Date => instant !== null, + ); + const { bands } = buildBandsForMedication({ + medication: input.medication, + schedule, + ctx, + userTz: input.userTz, + range: { from, to: at }, + now: input.now, + intakeInstants, + }); + for (const band of bands) { + if (band.at.getTime() !== at.getTime()) continue; + if (earliest === null || band.earlyStart.getTime() < earliest.getTime()) { + earliest = band.earlyStart; + } + } + } + + return earliest; +} diff --git a/src/lib/openapi/routes/insights/schemas.ts b/src/lib/openapi/routes/insights/schemas.ts index 28d6ebfc1..f9e1ce047 100644 --- a/src/lib/openapi/routes/insights/schemas.ts +++ b/src/lib/openapi/routes/insights/schemas.ts @@ -897,9 +897,20 @@ export const dashboardSnapshotResponse = z nextDueOverdue: z.boolean(), nextDueMedicationName: z.string().nullable(), nextDueMedicationId: z.string().nullable(), + dueCandidates: z + .array( + z.object({ + medicationId: z.string(), + medicationName: z.string(), + dueAt: z.string(), + overdue: z.boolean(), + availableFrom: z.string(), + }), + ) + .optional(), }) .describe( - "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), and the earliest next-due slot across active medications. `nextDueOverdue: true` marks an OPEN overdue slot (anchor passed, still inside its catch-up band). `nextDueMedicationId` deep-links a consumer straight to the overdue medication's card. Staleness contract: the snapshot is cache-served, so a `nextDueAt` in the past with `nextDueOverdue: false` means the anchor passed after the snapshot was built — render the plain day summary, never an overdue state.", + "Today's medication block: active-medication count, today-window tally (scheduled / taken / skipped), every active medication's current display-due candidate, and legacy scalar fields projected from candidate zero. `overdue: true` marks an OPEN overdue slot (anchor passed, still inside its catch-up band); `availableFrom` is the canonical cadence- and dose-window-derived attribution-band start. `dueCandidates` is optional so older cached snapshots remain valid. Medication ids let consumers deep-link straight to the relevant card.", ), // Dashboard hero — health score (warm phase, nullable on a // rollup-coverage miss). Score + band + delta only; the per-pillar diff --git a/tests/integration/apple-health-import-reconcile.test.ts b/tests/integration/apple-health-import-reconcile.test.ts index 541a42034..7dbd7c9a4 100644 --- a/tests/integration/apple-health-import-reconcile.test.ts +++ b/tests/integration/apple-health-import-reconcile.test.ts @@ -22,6 +22,7 @@ import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getPrismaClient, truncateAllTables } from "./setup"; import { reconcileOrphanImportJobs, + APPLE_HEALTH_IMPORT_PARSER_REVISION, _setWorkerPrismaForTests, } from "@/lib/jobs/apple-health-import-worker"; @@ -75,6 +76,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => { pgBossJobId: "boss-active-fresh", status: "parsing", uploadBytes: 100, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); bossMock.getJobById.mockResolvedValue({ @@ -98,6 +100,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => { pgBossJobId: "boss-gone", status: "upserting", uploadBytes: 100, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); bossMock.getJobById.mockResolvedValue(null); @@ -118,6 +121,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => { pgBossJobId: "boss-terminal", status: "parsing", uploadBytes: 100, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); bossMock.getJobById.mockResolvedValue({ @@ -140,6 +144,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => { pgBossJobId: "boss-stale-active", status: "parsing", uploadBytes: 100, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); await backdateHeartbeat(row.id, 45); @@ -163,6 +168,7 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => { pgBossJobId: "boss-none", status: "parsing", uploadBytes: 100, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }, }); bossMock.handle = null; // getGlobalBoss() → null @@ -173,6 +179,26 @@ describe("reconcileOrphanImportJobs — liveness gate (issue #486)", () => { expect(after!.status).toBe("failed"); }); + it("leaves a legacy-default revision-1 row to a revision-1 worker", async () => { + const prisma = getPrismaClient(); + const user = await createUser("recon-legacy-revision"); + const row = await prisma.importJob.create({ + data: { + userId: user.id, + pgBossJobId: "boss-legacy", + status: "parsing", + uploadBytes: 100, + }, + }); + + expect(row.parserRevision).toBe(1); + await reconcileOrphanImportJobs(); + + const after = await prisma.importJob.findUnique({ where: { id: row.id } }); + expect(after!.status).toBe("parsing"); + expect(bossMock.getJobById).not.toHaveBeenCalled(); + }); + it("leaves terminal rows untouched", async () => { const prisma = getPrismaClient(); const user = await createUser("recon-already-terminal"); @@ -197,7 +223,12 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => { // Mirrors the exact `findFirst` the kick-off routes run. async function dedupLookup(userId: string, sha: string) { return getPrismaClient().importJob.findFirst({ - where: { userId, uploadSha256: sha, status: { not: "failed" } }, + where: { + userId, + uploadSha256: sha, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + status: { not: "failed" }, + }, orderBy: { startedAt: "desc" }, }); } @@ -212,6 +243,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => { failureReason: "ENOENT: no such file or directory", uploadBytes: 100, uploadSha256: "sha-failed", + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, completedAt: new Date(), }, }); @@ -228,6 +260,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => { status: "done", uploadBytes: 100, uploadSha256: "sha-viable", + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, completedAt: new Date(), }, }); @@ -245,6 +278,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => { status: "failed", uploadBytes: 100, uploadSha256: "sha-mixed", + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, completedAt: new Date(), }, }); @@ -254,6 +288,7 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => { status: "done", uploadBytes: 100, uploadSha256: "sha-mixed", + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, completedAt: new Date(), }, }); @@ -261,4 +296,34 @@ describe("kick-off dedup — status-scoped lookup (issue #486)", () => { const hit = await dedupLookup(user.id, "sha-mixed"); expect(hit!.id).toBe(done.id); }); + it("allows a revision-1 archive to be reprocessed at revision 2", async () => { + const prisma = getPrismaClient(); + const user = await createUser("dedup-parser-revision"); + await prisma.importJob.create({ + data: { + userId: user.id, + status: "done", + uploadBytes: 100, + uploadSha256: "sha-parser-revision", + parserRevision: 1, + completedAt: new Date(), + }, + }); + + expect(await dedupLookup(user.id, "sha-parser-revision")).toBeNull(); + + const current = await prisma.importJob.create({ + data: { + userId: user.id, + status: "done", + uploadBytes: 100, + uploadSha256: "sha-parser-revision", + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + completedAt: new Date(), + }, + }); + expect((await dedupLookup(user.id, "sha-parser-revision"))?.id).toBe( + current.id, + ); + }); }); diff --git a/tests/integration/drain-per-sample-cumulative.test.ts b/tests/integration/drain-per-sample-cumulative.test.ts index aa4632a06..a7baf16cf 100644 --- a/tests/integration/drain-per-sample-cumulative.test.ts +++ b/tests/integration/drain-per-sample-cumulative.test.ts @@ -92,6 +92,10 @@ describe("drainPerSampleCumulative (real Postgres)", () => { expect(remaining[0].measuredAt.toISOString()).toBe( "2026-05-16T10:00:00.000Z", ); + expect(remaining[0]).toMatchObject({ + aggregationProvenance: "LEGACY_UNKNOWN", + syncVersion: 1, + }); }); it("is idempotent — a second run after a successful drain is a no-op", async () => { @@ -219,6 +223,93 @@ describe("drainPerSampleCumulative (real Postgres)", () => { expect(remaining).toHaveLength(2); }); + it.each(["HEALTHKIT_STATISTICS", "EXPORT_XML_SOURCE_MAX"] as const)( + "drains source rows without mutating an authoritative %s total", + async (aggregationProvenance) => { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 5_400, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-16T10:00:00.000Z"), + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16", + aggregationProvenance, + }, + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 500, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-16T20:00:00.000Z"), + externalId: `late-${aggregationProvenance}`, + }, + ], + }); + + await drainPerSampleCumulative(prisma, { + userId: TEST_USER_ID, + log: () => {}, + }); + + const remaining = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS" }, + }); + expect(remaining).toHaveLength(1); + expect(remaining[0]).toMatchObject({ + value: 5_400, + aggregationProvenance, + syncVersion: 1, + }); + }, + ); + + it("keeps legacy late-increment semantics and bumps syncVersion", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 5_400, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-16T10:00:00.000Z"), + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16", + aggregationProvenance: "LEGACY_UNKNOWN", + }, + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 500, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-16T20:00:00.000Z"), + externalId: "late-legacy", + }, + ], + }); + + await drainPerSampleCumulative(prisma, { + userId: TEST_USER_ID, + log: () => {}, + }); + + const remaining = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS" }, + }); + expect(remaining).toHaveLength(1); + expect(remaining[0]).toMatchObject({ + value: 5_900, + aggregationProvenance: "LEGACY_UNKNOWN", + syncVersion: 2, + }); + }); + it("skips rows already in stats:... shape (re-running after an earlier drain)", async () => { const prisma = getPrismaClient(); await prisma.measurement.create({ diff --git a/tests/integration/measurement-identity-reconcile.test.ts b/tests/integration/measurement-identity-reconcile.test.ts index 4835692cf..9ebfa33ba 100644 --- a/tests/integration/measurement-identity-reconcile.test.ts +++ b/tests/integration/measurement-identity-reconcile.test.ts @@ -97,6 +97,130 @@ describe("external measurement identity reconciliation (real Postgres)", () => { }); }); + it("retires a lower-authority XML collision while preserving the native aggregate", async () => { + const prisma = getPrismaClient(); + const user = await createUser("identity-native-authority"); + const nativeMeasuredAt = new Date("2026-07-20T10:00:00.000Z"); + const xmlExternal = await prisma.measurement.create({ + data: { + ...desired( + user.id, + "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + { + type: "ACTIVITY_STEPS", + value: 8_526, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T09:00:00.000Z"), + sleepStage: null, + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: 2, + aggregationSelectedSourceHash: "xml-source-hash", + }, + ), + }, + }); + const native = await prisma.measurement.create({ + data: { + ...desired(user.id, "stats:legacy-native-id", { + type: "ACTIVITY_STEPS", + value: 8_600, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: nativeMeasuredAt, + sleepStage: null, + externalSourceVersion: "native-v1", + deviceType: "Apple Watch", + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + }), + }, + }); + + const verdict = await reconcile( + desired(user.id, "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", { + type: "ACTIVITY_STEPS", + value: 8_526, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: nativeMeasuredAt, + sleepStage: null, + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: 2, + aggregationSelectedSourceHash: "xml-source-hash", + }), + { exactExternalMatch: "update" }, + ); + + expect(verdict).toMatchObject({ + status: "duplicate", + row: { id: native.id, externalId: "stats:legacy-native-id" }, + retiredCollisionId: xmlExternal.id, + }); + const liveRows = await prisma.measurement.findMany({ + where: { userId: user.id, deletedAt: null }, + }); + expect(liveRows).toHaveLength(1); + expect(liveRows[0]).toMatchObject({ + id: native.id, + value: 8_600, + unit: "steps", + measuredAt: nativeMeasuredAt, + externalId: "stats:legacy-native-id", + externalSourceVersion: "native-v1", + deviceType: "Apple Watch", + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + }); + expect( + await prisma.measurement.findUniqueOrThrow({ + where: { id: xmlExternal.id }, + }), + ).toMatchObject({ + externalId: `retired:${xmlExternal.id}:stats:HKQuantityTypeIdentifierStepCount:2026-07-20`, + deletedAt: expect.any(Date), + }); + + const nativeUpdate = await reconcile( + desired(user.id, "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", { + type: "ACTIVITY_STEPS", + value: 8_700, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: nativeMeasuredAt, + sleepStage: null, + externalSourceVersion: "native-v2", + deviceType: "Apple Watch", + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + }), + { exactExternalMatch: "update" }, + ); + + expect(nativeUpdate).toMatchObject({ + status: "updated", + row: { + id: native.id, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + }, + }); + expect( + await prisma.measurement.findMany({ + where: { userId: user.id, deletedAt: null }, + }), + ).toEqual([ + expect.objectContaining({ + id: native.id, + value: 8_700, + externalSourceVersion: "native-v2", + aggregationProvenance: "HEALTHKIT_STATISTICS", + }), + ]); + }); + it("adopts a live natural-key row when a new external id collides", async () => { const prisma = getPrismaClient(); const user = await createUser("identity-natural-live"); diff --git a/tests/integration/measurements-batch.test.ts b/tests/integration/measurements-batch.test.ts index cd2c24d40..d0e4c9ee7 100644 --- a/tests/integration/measurements-batch.test.ts +++ b/tests/integration/measurements-batch.test.ts @@ -9,10 +9,14 @@ * - Idempotency-Key replay returns the cached envelope */ import { NextRequest } from "next/server"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { cookieJar } from "./mock-next-headers"; import { getPrismaClient, truncateAllTables } from "./setup"; +import { streamParseExportXml } from "@/lib/measurements/import-apple-health-export"; const TEST_USER_ID = "user-batch-ingest-test"; @@ -89,6 +93,41 @@ function makeRequest( }); } +function writeCumulativeXml(firstValue = 8_526, secondValue = 7_082): string { + const tmp = mkdtempSync(join(tmpdir(), "healthlog-aggregate-authority-")); + const xmlPath = join(tmp, "export.xml"); + writeFileSync( + xmlPath, + ` + + + +`, + ); + return xmlPath; +} + +function nativeStepsRequest(value: number): NextRequest { + return makeRequest({ + entries: [ + { + hkIdentifier: "HKQuantityTypeIdentifierStepCount", + value, + unit: "count", + startDate: "2026-07-20T00:00:00.000Z", + endDate: "2026-07-20T18:00:00.000Z", + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + }, + ], + }); +} + describe("POST /api/measurements/batch (real Postgres)", () => { it("inserts a well-formed batch and returns per-entry status", async () => { const { POST } = await import("@/app/api/measurements/batch/route"); @@ -560,6 +599,12 @@ describe("POST /api/measurements/batch (real Postgres)", () => { }); expect(stored).toHaveLength(1); expect(stored[0]!.value).toBeCloseTo(5200, 5); + expect(stored[0]).toMatchObject({ + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + syncVersion: 2, + }); }); it("sample-class externalIds (non-stats:* prefix) keep the strict duplicate contract", async () => { @@ -1153,4 +1198,140 @@ describe("POST /api/measurements/batch (real Postgres)", () => { expect(stored).toHaveLength(0); }); }); + describe("Apple aggregate authority reconciliation", () => { + it("promotes an XML estimate to native HealthKit statistics", async () => { + const prisma = getPrismaClient(); + await streamParseExportXml({ + xmlPath: writeCumulativeXml(), + userId: TEST_USER_ID, + userTimezone: "Europe/Berlin", + prisma, + }); + const estimated = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + }, + }); + expect(estimated).toHaveLength(1); + expect(estimated[0]).toMatchObject({ + value: 8_526, + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: 2, + syncVersion: 1, + }); + + const { POST } = await import("@/app/api/measurements/batch/route"); + const response = await POST(nativeStepsRequest(8_600)); + expect(response.status).toBe(200); + + const canonical = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + }, + }); + expect(canonical).toHaveLength(1); + expect(canonical[0]).toMatchObject({ + value: 8_600, + aggregationProvenance: "HEALTHKIT_STATISTICS", + aggregationContributorCount: null, + aggregationSelectedSourceHash: null, + syncVersion: 2, + }); + }); + + it("keeps native statistics authoritative when XML arrives later", async () => { + const { POST } = await import("@/app/api/measurements/batch/route"); + expect((await POST(nativeStepsRequest(8_600))).status).toBe(200); + + const result = await streamParseExportXml({ + xmlPath: writeCumulativeXml(), + userId: TEST_USER_ID, + userTimezone: "Europe/Berlin", + prisma: getPrismaClient(), + }); + expect(result.cumulativeEstimates).toEqual({ days: 1, rows: 1 }); + expect(result.perType.ACTIVITY_STEPS).toMatchObject({ + inserted: 0, + updated: 0, + }); + + const canonical = await getPrismaClient().measurement.findMany({ + where: { + userId: TEST_USER_ID, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + }, + }); + expect(canonical).toHaveLength(1); + expect(canonical[0]).toMatchObject({ + value: 8_600, + aggregationProvenance: "HEALTHKIT_STATISTICS", + syncVersion: 1, + }); + }); + + it("re-imports an XML estimate onto one row with monotonic syncVersion", async () => { + const prisma = getPrismaClient(); + const xmlPath = writeCumulativeXml(); + await streamParseExportXml({ + xmlPath, + userId: TEST_USER_ID, + userTimezone: "Europe/Berlin", + prisma, + }); + await streamParseExportXml({ + xmlPath, + userId: TEST_USER_ID, + userTimezone: "Europe/Berlin", + prisma, + }); + + const rows = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + }, + }); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + value: 8_526, + aggregationProvenance: "EXPORT_XML_SOURCE_MAX", + aggregationContributorCount: 2, + syncVersion: 2, + }); + }); + + it("repairs an explicitly legacy aggregate with native statistics", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 15_608, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-07-20T10:00:00.000Z"), + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + aggregationProvenance: "LEGACY_UNKNOWN", + }, + }); + + const { POST } = await import("@/app/api/measurements/batch/route"); + expect((await POST(nativeStepsRequest(8_600))).status).toBe(200); + + const repaired = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-07-20", + }, + }); + expect(repaired).toHaveLength(1); + expect(repaired[0]).toMatchObject({ + value: 8_600, + aggregationProvenance: "HEALTHKIT_STATISTICS", + syncVersion: 2, + }); + }); + }); });