diff --git a/CHANGELOG.md b/CHANGELOG.md index 23cd87923..297cbf8d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,18 @@ an earlier fold computed from part of a day. highest pollen join the correlation channels. An account switch under Location & weather (on by default), attributions beside the values, and a dashboard chip on a day with high pollen, a warm night or very poor air. +- **Air quality for the past.** A background job fills air quality, pollen + and UV for earlier local days with an entry: after the update, when the + home is set and when air quality is switched on. A day inside a location + period is fetched at that place, a day from the home's effective date on + at the home; a day before it with no period stays empty rather than + borrowing a place the person was not at. The source reaches back to 2013 + in Europe and to August 2022 elsewhere, pollen from 2021 in Europe only; + older days stay empty. Nearby days share one + request of up to 90 days, a run holds itself to 300 of the account's 400 + daily calls and half of each instance window, and a long history spreads + over several nights. Location & weather shows the progress (migration + 0391). - **`get_environment` and `get_day`** for the Coach and MCP. `get_environment` is module-gated and carries no coordinate or place name; `get_day` leaves out life events, notes and document names and honours @@ -77,8 +89,10 @@ an earlier fold computed from part of a day. when each type last arrived, by which trigger (including the new `manual` for Sync all) and when it last brought a new value (migration 0384). -- **Documents layout.** Preview cards or a compact list, months stacked or - flowing, per account (migration 0388). +- **Documents layout.** Preview cards or a compact list, months flowing + side by side (the default) or stacked, per account (migration 0388). Only + a field the person picked is stored, so an untouched one follows the + default. - **Measurement maintenance.** An admin card and `POST /api/admin/maintenance/measurements` run `VACUUM (ANALYZE)` and a `REINDEX INDEX CONCURRENTLY` per index, largest first. @@ -180,7 +194,8 @@ All changes are additive. `lastNewSampleAt` per type. - The sync feed lists only deletions a person makes, not those of the condensing step. -- `GET /api/environment` adds `airQuality`, `latestDay` and `attributions`. +- `GET /api/environment` adds `airQuality` (with `history`, the progress of + the past-days backfill), `latestDay` and `attributions`. - `GET`/`PUT /api/auth/me/source-priority` add `inUse`. - A dose for a record-only medication is refused with 422 `medication.intake.notTracked`; failed passkey sign-ins answer with diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index d17709fd3..54127a1dd 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -71396,7 +71396,7 @@ components: - stacked - flow description: '`stacked` gives every month its own block; `flow` runs the documents across the full width and wraps, with - month names inline. Default "stacked".' + month names inline. Default "flow"; an account that never chose keeps following the default.' required: - version - view @@ -79630,12 +79630,43 @@ components: - type: "null" description: "The model domain of that day: `cams_europe` (about 11 km, with pollen) or `cams_global` (about 45 km, no pollen)." + history: + anyOf: + - type: object + properties: + total: + type: integer + minimum: -9007199254740991 + maximum: 9007199254740991 + description: Past days with entries the source can serve at their place (from 2013 in Europe, from August 2022 + elsewhere). Older days stay without values. + done: + type: integer + minimum: -9007199254740991 + maximum: 9007199254740991 + description: Of those, the days whose air quality is stored. + complete: + type: boolean + description: True when nothing the source serves is left. + checkedAt: + type: string + description: ISO instant of the run that counted them. + required: + - total + - done + - complete + - checkedAt + additionalProperties: false + - type: "null" + description: Progress of the background backfill that fills air quality for every past day with entries. Null before its + first run and while air quality is off. required: - enabled - operatorDisabled - days - latestDate - domain + - history additionalProperties: false description: v1.42 — the air-quality part of the module. latestDay: diff --git a/messages/de.json b/messages/de.json index 85f28b659..595f87d0b 100644 --- a/messages/de.json +++ b/messages/de.json @@ -5829,6 +5829,8 @@ "operatorDisabled": "Luftqualität ist auf diesem Server abgeschaltet.", "coverage": "{count} Tage mit Luftqualitätsdaten", "pollenEuropeOnly": "Pollendaten gibt es nur für Europa.", + "historyProgress": "Luftqualität wird nachgeladen: {done} von {total} Tagen", + "historyComplete": "Luftqualität ist für alle früheren Tage mit Einträgen nachgeladen, die die Quelle abdeckt ({total} Tage). Ältere Tage bleiben leer.", "attribution": "Datenquellen und Lizenzen", "saveError": "Die Einstellung konnte nicht gespeichert werden.", "latest": "Jüngster Tag ({date})", diff --git a/messages/en.json b/messages/en.json index ef3b5feba..452504d86 100644 --- a/messages/en.json +++ b/messages/en.json @@ -5829,6 +5829,8 @@ "operatorDisabled": "Air quality is turned off on this server.", "coverage": "{count} days with air-quality data", "pollenEuropeOnly": "Pollen data covers Europe only.", + "historyProgress": "Backfilling air quality: {done} of {total} days", + "historyComplete": "Air quality is filled in for every earlier day with entries that the source covers ({total} days). Older days stay empty.", "attribution": "Data sources and licences", "saveError": "The setting could not be saved.", "latest": "Newest day ({date})", diff --git a/messages/es.json b/messages/es.json index b920dd20c..4eeead65f 100644 --- a/messages/es.json +++ b/messages/es.json @@ -5829,6 +5829,8 @@ "operatorDisabled": "La calidad del aire está desactivada en este servidor.", "coverage": "{count} días con datos de calidad del aire", "pollenEuropeOnly": "Los datos de polen solo cubren Europa.", + "historyProgress": "Completando la calidad del aire: {done} de {total} días", + "historyComplete": "La calidad del aire está completada para todos los días anteriores con registros que cubre la fuente ({total} días). Los días más antiguos quedan vacíos.", "attribution": "Fuentes de datos y licencias", "saveError": "No se pudo guardar el ajuste.", "latest": "Día más reciente ({date})", diff --git a/messages/fr.json b/messages/fr.json index 37ab610cd..f7e73f1f6 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -5829,6 +5829,8 @@ "operatorDisabled": "La qualité de l’air est désactivée sur ce serveur.", "coverage": "{count} jours avec des données de qualité de l’air", "pollenEuropeOnly": "Les données de pollen ne couvrent que l’Europe.", + "historyProgress": "Récupération de la qualité de l’air : {done} jours sur {total}", + "historyComplete": "La qualité de l’air est complétée pour tous les jours passés avec des saisies que la source couvre ({total} jours). Les jours plus anciens restent vides.", "attribution": "Sources de données et licences", "saveError": "Le réglage n’a pas pu être enregistré.", "latest": "Jour le plus récent ({date})", diff --git a/messages/it.json b/messages/it.json index a763be6bf..8f609b89a 100644 --- a/messages/it.json +++ b/messages/it.json @@ -5829,6 +5829,8 @@ "operatorDisabled": "La qualità dell’aria è disattivata su questo server.", "coverage": "{count} giorni con dati sulla qualità dell’aria", "pollenEuropeOnly": "I dati sui pollini coprono solo l’Europa.", + "historyProgress": "Recupero della qualità dell’aria: {done} di {total} giorni", + "historyComplete": "La qualità dell’aria è completata per tutti i giorni passati con registrazioni coperti dalla fonte ({total} giorni). I giorni più vecchi restano vuoti.", "attribution": "Fonti dei dati e licenze", "saveError": "Impossibile salvare l’impostazione.", "latest": "Giorno più recente ({date})", diff --git a/messages/ko.json b/messages/ko.json index 42938c942..d5d5b337d 100644 --- a/messages/ko.json +++ b/messages/ko.json @@ -5829,6 +5829,8 @@ "operatorDisabled": "이 서버에서는 대기질 기능이 꺼져 있습니다.", "coverage": "대기질 데이터가 있는 날 {count}일", "pollenEuropeOnly": "꽃가루 데이터는 유럽만 제공됩니다.", + "historyProgress": "대기질 과거 데이터 채우는 중: {total}일 중 {done}일", + "historyComplete": "출처가 제공하는 기록이 있는 이전 날짜 모두에 대기질이 채워졌습니다({total}일). 그보다 오래된 날짜는 비어 있습니다.", "attribution": "데이터 출처와 라이선스", "saveError": "설정을 저장하지 못했습니다.", "latest": "가장 최근 날짜 ({date})", diff --git a/messages/pl.json b/messages/pl.json index 56627bedc..395396400 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -5829,6 +5829,8 @@ "operatorDisabled": "Jakość powietrza jest wyłączona na tym serwerze.", "coverage": "Dni z danymi o jakości powietrza: {count}", "pollenEuropeOnly": "Dane o pyłkach obejmują tylko Europę.", + "historyProgress": "Uzupełnianie jakości powietrza: {done} z {total} dni", + "historyComplete": "Jakość powietrza uzupełniono dla wszystkich wcześniejszych dni z wpisami, które obejmuje źródło ({total} dni). Starsze dni pozostają puste.", "attribution": "Źródła danych i licencje", "saveError": "Nie udało się zapisać ustawienia.", "latest": "Najnowszy dzień ({date})", diff --git a/prisma/migrations/0391_environment_aq_history/migration.sql b/prisma/migrations/0391_environment_aq_history/migration.sql new file mode 100644 index 000000000..67b503977 --- /dev/null +++ b/prisma/migrations/0391_environment_aq_history/migration.sql @@ -0,0 +1,4 @@ +-- Progress of the air-quality history backfill (days covered of days the +-- source can serve). Written by the background job only; null until its +-- first run. +ALTER TABLE "users" ADD COLUMN "environment_aq_history_json" JSONB; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index a9118035e..911083311 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -113,6 +113,12 @@ model User { /// nightly fetch skips the air-quality feed and the surfaces leave the /// fields out. The operator switch `ENVIRONMENT_AIR_QUALITY_DISABLED` wins. environmentAirQualityEnabled Boolean @default(true) @map("environment_air_quality_enabled") + /// v1.42 — progress of the air-quality history backfill + /// (`src/lib/environment/air-quality-history.ts`): how many past days with + /// entries the source can cover, how many carry air quality, when it was + /// last checked and whether it is through. Written by the job only; read by + /// `GET /api/environment` for the settings card. Null until the first run. + environmentAqHistoryJson Json? @map("environment_aq_history_json") // Codex OAuth (ChatGPT subscription) codexAccessTokenEncrypted String? @map("codex_access_token_encrypted") diff --git a/src/app/api/documents/inbound/layout/__tests__/route.test.ts b/src/app/api/documents/inbound/layout/__tests__/route.test.ts index 8c267af61..41ca79b2b 100644 --- a/src/app/api/documents/inbound/layout/__tests__/route.test.ts +++ b/src/app/api/documents/inbound/layout/__tests__/route.test.ts @@ -89,9 +89,27 @@ describe("GET /api/documents/inbound/layout", () => { const res = await callGet(); expect(res.status).toBe(200); expect((await res.json()).data).toEqual(DEFAULT_DOCUMENTS_LAYOUT); + expect(DEFAULT_DOCUMENTS_LAYOUT).toEqual({ + version: 1, + view: "cards", + arrangement: "flow", + }); expect(prisma.user.update).not.toHaveBeenCalled(); }); + it("keeps a saved stacked arrangement although the default is flow", async () => { + vi.mocked(prisma.user.findUnique).mockResolvedValue({ + documentsLayoutJson: { + version: 1, + view: "cards", + arrangement: "stacked", + }, + } as never); + + const res = await callGet(); + expect((await res.json()).data.arrangement).toBe("stacked"); + }); + it("serves the stored choice, repairing an unknown value field by field", async () => { vi.mocked(prisma.user.findUnique).mockResolvedValue({ documentsLayoutJson: { version: 1, view: "list", arrangement: "spiral" }, @@ -101,7 +119,7 @@ describe("GET /api/documents/inbound/layout", () => { expect((await res.json()).data).toEqual({ version: 1, view: "list", - arrangement: "stacked", + arrangement: "flow", }); }); @@ -146,6 +164,23 @@ describe("PUT /api/documents/inbound/layout", () => { }); }); + it("stores only the chosen field, so the untouched one keeps following the default", async () => { + vi.mocked(prisma.user.findUnique).mockResolvedValue({ + documentsLayoutJson: null, + } as never); + + const res = await put({ version: 1, view: "list" }); + expect((await res.json()).data).toEqual({ + version: 1, + view: "list", + arrangement: "flow", + }); + expect(prisma.user.update).toHaveBeenCalledWith({ + where: { id: "user-1" }, + data: { documentsLayoutJson: { version: 1, view: "list" } }, + }); + }); + it("refuses an unknown value and an unknown key with 422 and writes nothing", async () => { const bad = await put({ version: 1, view: "carousel" }); expect(bad.status).toBe(422); diff --git a/src/app/api/documents/inbound/layout/route.ts b/src/app/api/documents/inbound/layout/route.ts index 45d3e2b94..e81e5bca8 100644 --- a/src/app/api/documents/inbound/layout/route.ts +++ b/src/app/api/documents/inbound/layout/route.ts @@ -20,6 +20,7 @@ import { DOCUMENTS_LAYOUT_ARRANGEMENTS, DOCUMENTS_LAYOUT_VIEWS, resolveDocumentsLayout, + storedDocumentsLayoutChoices, type DocumentsLayout, } from "@/lib/documents/documents-layout"; import { annotate } from "@/lib/logging/context"; @@ -35,12 +36,12 @@ const layoutPutSchema = z }) .strict(); -async function readLayout(userId: string): Promise { +async function readStoredLayout(userId: string): Promise { const row = await prisma.user.findUnique({ where: { id: userId }, select: { documentsLayoutJson: true }, }); - return resolveDocumentsLayout(row?.documentsLayoutJson); + return row?.documentsLayoutJson; } export const GET = apiHandler(async () => { @@ -51,7 +52,7 @@ export const GET = apiHandler(async () => { const gate = await requireModuleEnabled(user.id, "inboundDocuments"); if (!gate.enabled) return gate.response; - const layout = await readLayout(user.id); + const layout = resolveDocumentsLayout(await readStoredLayout(user.id)); annotate({ action: { name: "documents.layout.read" }, meta: { view: layout.view, arrangement: layout.arrangement }, @@ -78,17 +79,20 @@ export const PUT = apiHandler(async (request: Request) => { } // Preserve-when-absent: one stored read fills whichever field the body - // left out. - const existing = await readLayout(user.id); - const next: DocumentsLayout = { - version: 1, - view: parsed.data.view ?? existing.view, - arrangement: parsed.data.arrangement ?? existing.arrangement, + // left out. Only fields the user has chosen are written, so a field never + // picked keeps following the default (which can change between releases). + const choices = { + ...storedDocumentsLayoutChoices(await readStoredLayout(user.id)), + ...(parsed.data.view ? { view: parsed.data.view } : {}), + ...(parsed.data.arrangement + ? { arrangement: parsed.data.arrangement } + : {}), }; + const next: DocumentsLayout = resolveDocumentsLayout(choices); await prisma.user.update({ where: { id: user.id }, - data: { documentsLayoutJson: toJson(next) }, + data: { documentsLayoutJson: toJson({ version: 1, ...choices }) }, }); annotate({ diff --git a/src/app/api/environment/home/route.ts b/src/app/api/environment/home/route.ts index f6686a6c3..3c6d554ce 100644 --- a/src/app/api/environment/home/route.ts +++ b/src/app/api/environment/home/route.ts @@ -18,6 +18,7 @@ import { homeLocationSchema } from "@/lib/validations/environment"; import { roundCoarse } from "@/lib/environment/open-meteo"; import { sealLocation } from "@/lib/environment/location-cipher"; import { enqueueEnvironmentFetch } from "@/lib/jobs/environment-fetch"; +import { enqueueAirQualityHistory } from "@/lib/jobs/environment-air-quality-history"; export const PUT = apiHandler(async (request: NextRequest) => { const { user } = await requireAuth(); @@ -67,6 +68,9 @@ export const PUT = apiHandler(async (request: NextRequest) => { // Kick a lookback refresh so recent days populate promptly. No-ops cleanly // when no worker is bound; the nightly cron still covers it. await enqueueEnvironmentFetch({ userId: user.id }); + // And the past: air quality for every earlier day with entries, in the + // background and inside the request budget. A no-op while it is off. + await enqueueAirQualityHistory(user.id); annotate({ action: { name: "environment.home.set" } }); diff --git a/src/app/api/environment/preferences/route.ts b/src/app/api/environment/preferences/route.ts index 578f142c0..e8e30f43b 100644 --- a/src/app/api/environment/preferences/route.ts +++ b/src/app/api/environment/preferences/route.ts @@ -20,6 +20,7 @@ import { prisma } from "@/lib/db"; import { environmentPreferencesSchema } from "@/lib/validations/environment"; import { isAirQualityOperatorDisabled } from "@/lib/environment/open-meteo-air-quality"; import { enqueueEnvironmentFetch } from "@/lib/jobs/environment-fetch"; +import { enqueueAirQualityHistory } from "@/lib/jobs/environment-air-quality-history"; export const PATCH = apiHandler(async (request: NextRequest) => { const { user } = await requireAuth(); @@ -57,6 +58,8 @@ export const PATCH = apiHandler(async (request: NextRequest) => { before?.environmentAirQualityEnabled === false ) { await enqueueEnvironmentFetch({ userId: user.id }); + // Switched on: fill the past as well, every earlier day with entries. + await enqueueAirQualityHistory(user.id); } annotate({ diff --git a/src/app/api/environment/route.ts b/src/app/api/environment/route.ts index 1a4520daf..cb75a7260 100644 --- a/src/app/api/environment/route.ts +++ b/src/app/api/environment/route.ts @@ -11,7 +11,8 @@ * (the account switch, the operator switch, how many days carry air-quality * values), the newest stored day for the dashboard chips (with its air * quality only while the part is on), and every attribution line in - * `attributions`. + * `attributions`, and the progress of the air-quality history backfill + * (`airQuality.history`). */ import { apiHandler, requireAuth } from "@/lib/api-handler"; import { apiSuccess } from "@/lib/api-response"; @@ -23,6 +24,7 @@ import { isAirQualityOperatorDisabled } from "@/lib/environment/open-meteo-air-q import { environmentAttributionLines } from "@/lib/environment/air-quality-contract"; import { readLocation } from "@/lib/environment/location-cipher"; import { isAirQualityActive } from "@/lib/environment/service"; +import { readAirQualityHistoryState } from "@/lib/environment/air-quality-history"; import { ENVIRONMENT_FETCH_QUEUE } from "@/lib/jobs/environment-fetch"; import { readQueueFailureForUser } from "@/lib/jobs/job-failures"; @@ -54,6 +56,7 @@ export const GET = apiHandler(async () => { homeSince: true, timezone: true, environmentAirQualityEnabled: true, + environmentAqHistoryJson: true, }, }), prisma.environmentTravelLocation.findMany({ @@ -157,6 +160,19 @@ export const GET = apiHandler(async () => { const accountEnabled = profile?.environmentAirQualityEnabled ?? true; const operatorDisabled = isAirQualityOperatorDisabled(); const airActive = isAirQualityActive(accountEnabled); + // The history backfill's progress, while the part is on: what the last run + // counted, so the card can say how far the past has been filled. + const historyState = airActive + ? readAirQualityHistoryState(profile?.environmentAqHistoryJson) + : null; + const history = historyState + ? { + total: historyState.total, + done: historyState.done, + complete: historyState.complete, + checkedAt: historyState.checkedAt, + } + : null; const latestYear = Number( (latestAir?.date ?? latest?.date ?? new Date().toISOString()).slice(0, 4), ); @@ -219,6 +235,7 @@ export const GET = apiHandler(async () => { days: airDays, latestDate: latestAir?.date ?? null, domain: latestAir?.aqDomain ?? null, + history, }, latestDay, // null = the last background runs did not fail (or there is no queue to diff --git a/src/components/documents/document-timeline.tsx b/src/components/documents/document-timeline.tsx index d8eb507b7..a1b75597a 100644 --- a/src/components/documents/document-timeline.tsx +++ b/src/components/documents/document-timeline.tsx @@ -31,9 +31,10 @@ import { useVirtualizer } from "@tanstack/react-virtual"; import { useTranslations } from "@/lib/i18n/context"; import type { InboundDocumentDto } from "@/lib/validations/inbound-documents"; import { UploadStateCard } from "./document-card"; -import type { - DocumentsLayoutArrangement, - DocumentsLayoutView, +import { + DEFAULT_DOCUMENTS_LAYOUT, + type DocumentsLayoutArrangement, + type DocumentsLayoutView, } from "@/lib/documents/documents-layout"; import { columnsForWidth, @@ -65,8 +66,8 @@ export function DocumentTimeline({ highlightId, onPrefetch, timezone, - view = "cards", - arrangement = "stacked", + view = DEFAULT_DOCUMENTS_LAYOUT.view, + arrangement = DEFAULT_DOCUMENTS_LAYOUT.arrangement, }: { /** Preview tiles or compact rows (the reader's vault presentation). */ view?: DocumentsLayoutView; diff --git a/src/components/settings/environment-section.tsx b/src/components/settings/environment-section.tsx index 69c27566e..58532e5d3 100644 --- a/src/components/settings/environment-section.tsx +++ b/src/components/settings/environment-section.tsx @@ -92,6 +92,13 @@ interface EnvironmentOverview { days: number; latestDate: string | null; domain: string | null; + /** The history backfill's progress; null before its first run. */ + history: { + total: number; + done: number; + complete: boolean; + checkedAt: string; + } | null; }; /** v1.42 — the newest stored day; `airQuality` null while off/unfetched. */ latestDay: { @@ -638,6 +645,26 @@ export function EnvironmentSection() { ? ` ${t("settings.sections.environment.airQuality.pollenEuropeOnly")}` : ""}

+ {data.airQuality.history && data.airQuality.history.total > 0 && ( +

+ {data.airQuality.history.complete + ? t( + "settings.sections.environment.airQuality.historyComplete", + { total: data.airQuality.history.total }, + ) + : t( + "settings.sections.environment.airQuality.historyProgress", + { + done: data.airQuality.history.done, + total: data.airQuality.history.total, + }, + )} +

+ )} {data.latestDay?.airQuality && (

diff --git a/src/lib/data-wipe/wipe-plan.ts b/src/lib/data-wipe/wipe-plan.ts index df6254c80..d0cd9958a 100644 --- a/src/lib/data-wipe/wipe-plan.ts +++ b/src/lib/data-wipe/wipe-plan.ts @@ -378,6 +378,7 @@ export const USER_RESET = { homeSince: null, homeLocationEncrypted: null, environmentAirQualityEnabled: true, + environmentAqHistoryJson: Prisma.DbNull, // Derived AI output cached on the row insightsPrivacyMode: "aggregated", diff --git a/src/lib/documents/documents-layout.ts b/src/lib/documents/documents-layout.ts index dc796ecd9..fcef2266d 100644 --- a/src/lib/documents/documents-layout.ts +++ b/src/lib/documents/documents-layout.ts @@ -10,9 +10,10 @@ * and wrap, with each month's name riding inline as a small marker that does * not break the run. * - * Null / missing column = the defaults below (cards, stacked — the vault as it - * has always looked). The GET endpoint never lazy-writes, so the column only - * carries data once the user has changed one of the two. Mirrors + * Null / missing column = the defaults below (cards, flowing months). The GET + * endpoint never lazy-writes, and the PUT stores only the fields the user has + * actually chosen, so a default that changes later reaches every account that + * never picked that field, while a saved choice stays. Mirrors * `medication-list-layout.ts`: tolerant resolver on read, normaliser on write, * preserve-when-absent PUT (see the route). */ @@ -33,7 +34,7 @@ export interface DocumentsLayout { export const DEFAULT_DOCUMENTS_LAYOUT: DocumentsLayout = { version: 1, view: "cards", - arrangement: "stacked", + arrangement: "flow", }; function isView(value: unknown): value is DocumentsLayoutView { @@ -68,3 +69,22 @@ export function resolveDocumentsLayout(raw: unknown): DocumentsLayout { : DEFAULT_DOCUMENTS_LAYOUT.arrangement, }; } + +/** + * The fields a stored blob explicitly carries, each validated. Unknown or + * malformed values drop out, so the default fills them on read. The PUT + * merges into this rather than into the resolved layout, which keeps an + * untouched field following the default instead of freezing today's value. + */ +export function storedDocumentsLayoutChoices( + raw: unknown, +): Partial> { + if (raw === null || raw === undefined || typeof raw !== "object") return {}; + const blob = raw as { view?: unknown; arrangement?: unknown }; + return { + ...(isView(blob.view) ? { view: blob.view } : {}), + ...(isArrangement(blob.arrangement) + ? { arrangement: blob.arrangement } + : {}), + }; +} diff --git a/src/lib/environment/__tests__/air-quality-history.test.ts b/src/lib/environment/__tests__/air-quality-history.test.ts new file mode 100644 index 000000000..4b58cccb4 --- /dev/null +++ b/src/lib/environment/__tests__/air-quality-history.test.ts @@ -0,0 +1,215 @@ +/** + * v1.42 — the air-quality history backfill's pure parts: how pending days + * are cut into request ranges, what a range weighs, the ceilings that keep + * the history from taking the nightly fetch's room, where a day without a + * row is placed, and when an account is offered a run again. + */ +import { describe, expect, it } from "vitest"; + +import { + AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS, + historyRangeWeight, + historyRequestFits, + planHistoryRanges, + readAirQualityHistoryState, + resolveHistoryLocation, +} from "../air-quality-history"; +import { historyDue } from "@/lib/jobs/environment-air-quality-history"; +import { shiftDateKey } from "@/lib/tz/format"; + +function daysFrom(start: string, count: number, step = 1): string[] { + return Array.from({ length: count }, (_, i) => shiftDateKey(start, i * step)); +} + +describe("planHistoryRanges", () => { + it("joins days up to 14 apart into one range and splits on a longer gap", () => { + const ranges = planHistoryRanges([ + "2024-01-01", + "2024-01-15", // 14 days later: joined + "2024-01-30", // 15 days later: new range + "2024-01-31", + ]); + expect(ranges.map((r) => [r.startDate, r.endDate, r.days.length])).toEqual([ + ["2024-01-01", "2024-01-15", 2], + ["2024-01-30", "2024-01-31", 2], + ]); + }); + + it("never lets a range grow past 90 days", () => { + const ranges = planHistoryRanges(daysFrom("2023-01-01", 200)); + expect(ranges).toHaveLength(3); + for (const r of ranges) { + expect(shiftDateKey(r.startDate, 89) >= r.endDate).toBe(true); + } + expect(ranges.flatMap((r) => r.days)).toHaveLength(200); + }); + + it("sorts and de-duplicates its input", () => { + const ranges = planHistoryRanges([ + "2024-03-02", + "2024-03-01", + "2024-03-02", + ]); + expect(ranges).toEqual([ + { + startDate: "2024-03-01", + endDate: "2024-03-02", + days: ["2024-03-01", "2024-03-02"], + }, + ]); + }); + + it("spends fewer calls on a sparse history than one request per day", () => { + // A day every three days for a year: 122 days. + const days = daysFrom("2023-01-01", 122, 3); + const ranges = planHistoryRanges(days); + const ranged = ranges.reduce( + (sum, r) => sum + historyRangeWeight(r, []), + 0, + ); + const perDay = + days.length * + historyRangeWeight( + { startDate: days[0], endDate: days[0], days: [days[0]] }, + [], + ); + expect(ranged).toBeLessThan(perDay / 4); + }); +}); + +describe("historyRangeWeight", () => { + it("adds the weather only over the span of the days without a row", () => { + const range = { + startDate: "2023-01-01", + endDate: "2023-03-31", + days: ["2023-01-01", "2023-03-31"], + }; + const airOnly = historyRangeWeight(range, []); + expect(airOnly).toBeCloseTo((1.7 * 90) / 14); + // One missing day: one weather request at its minimum weight (12 vars). + expect(historyRangeWeight(range, ["2023-03-31"])).toBeCloseTo( + airOnly + 1.2, + ); + }); +}); + +describe("historyRequestFits", () => { + it("admits a request under every ceiling", () => { + expect(historyRequestFits({}, 10)).toBe(true); + }); + + it("holds the account to three quarters of its daily share", () => { + expect(AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS).toBe(300); + expect(historyRequestFits({ "account-day": 295 }, 5)).toBe(true); + expect(historyRequestFits({ "account-day": 295 }, 5.1)).toBe(false); + }); + + it("holds the history to half of each instance window", () => { + expect(historyRequestFits({ minute: 245 }, 5)).toBe(true); + expect(historyRequestFits({ minute: 246 }, 5)).toBe(false); + expect(historyRequestFits({ hour: 1999 }, 2)).toBe(false); + expect(historyRequestFits({ day: 3995 }, 10)).toBe(false); + }); +}); + +describe("resolveHistoryLocation", () => { + const home = { + lat: 51.5, + lon: 7.2, + label: "Home", + timezone: "Europe/Berlin", + since: "2026-09-01", + }; + const trip = { + startDate: "2024-07-01", + endDate: "2024-07-14", + lat: 38.7, + lon: -9.1, + label: "Lisbon", + }; + + it("places a day of a trip at the trip", () => { + expect(resolveHistoryLocation("2024-07-05", home, [trip])).toMatchObject({ + lat: 38.7, + source: "TRAVEL", + }); + }); + + it("places a day from the home's effective date on at the home", () => { + expect(resolveHistoryLocation("2026-09-01", home, [trip])).toMatchObject({ + lat: 51.5, + source: "HOME", + }); + }); + + it("leaves a day before the home was set without a place, unless a period covers it", () => { + expect(resolveHistoryLocation("2026-08-31", home, [trip])).toBeNull(); + expect(resolveHistoryLocation("2020-03-01", home, [trip])).toBeNull(); + expect(resolveHistoryLocation("2024-07-14", home, [trip])).toMatchObject({ + source: "TRAVEL", + }); + }); + + it("places every day at a home stored without an effective date", () => { + expect( + resolveHistoryLocation("2015-01-01", { ...home, since: null }, []), + ).toMatchObject({ lat: 51.5, source: "HOME" }); + }); + + it("places nothing without a home or a period", () => { + expect(resolveHistoryLocation("2020-03-01", null, [trip])).toBeNull(); + }); +}); + +describe("stored progress", () => { + const now = new Date("2026-10-09T03:00:00Z"); + + it("reads a valid state and caps done at total", () => { + expect( + readAirQualityHistoryState({ + version: 1, + total: 10, + done: 12, + checkedAt: now.toISOString(), + complete: false, + }), + ).toMatchObject({ total: 10, done: 10 }); + expect(readAirQualityHistoryState({ total: "x" })).toBeNull(); + expect(readAirQualityHistoryState(null)).toBeNull(); + }); + + it("offers an account with no, an unfinished or a stale history", () => { + expect(historyDue(null, now)).toBe(true); + expect( + historyDue( + { total: 5, done: 2, checkedAt: now.toISOString(), complete: false }, + now, + ), + ).toBe(true); + expect( + historyDue( + { + total: 5, + done: 5, + checkedAt: "2026-09-30T03:00:00Z", + complete: true, + }, + now, + ), + ).toBe(true); + }); + + it("leaves a history that is through and checked this week alone", () => { + expect( + historyDue( + { + total: 5, + done: 5, + checkedAt: "2026-10-07T03:00:00Z", + complete: true, + }, + now, + ), + ).toBe(false); + }); +}); diff --git a/src/lib/environment/__tests__/open-meteo-air-quality.test.ts b/src/lib/environment/__tests__/open-meteo-air-quality.test.ts index d9fe7b580..d3f7db5d6 100644 --- a/src/lib/environment/__tests__/open-meteo-air-quality.test.ts +++ b/src/lib/environment/__tests__/open-meteo-air-quality.test.ts @@ -29,6 +29,8 @@ import { ozoneMax8h, uvIndexMax, variablesForChunk, + airQualityChunkWeight, + earliestAirQualityDay, type AirQualityHourly, } from "../open-meteo-air-quality"; @@ -172,14 +174,24 @@ describe("aggregateAirQuality", () => { }); describe("requests", () => { - it("splits into 90-day chunks and asks the pre-2022 ones for the pollutants only", () => { + it("splits into 90-day chunks and asks each era only for what it carries", () => { expect(chunkDays("2022-01-01", "2022-07-01")).toEqual([ { startDate: "2022-01-01", endDate: "2022-03-31" }, { startDate: "2022-04-01", endDate: "2022-06-29" }, { startDate: "2022-06-30", endDate: "2022-07-01" }, ]); - expect(variablesForChunk("2022-03-31")).toHaveLength(8); - expect(variablesForChunk("2022-06-29")).toHaveLength(17); + // Pollutants and indices alone before 2021. + expect(variablesForChunk("2020-12-31")).toHaveLength(8); + // Pollen and dust join in 2021 (Europe). + expect(variablesForChunk("2021-01-01")).toHaveLength(15); + expect(variablesForChunk("2022-07-31")).toHaveLength(15); + // UV and aerosol depth from CAMS global, August 2022 on. + expect(variablesForChunk("2022-08-01")).toHaveLength(17); + // Weight: 90 days of the full set is 1.7 × 90 / 14 calls. + expect(airQualityChunkWeight("2023-01-01", "2023-03-31")).toBeCloseTo( + (1.7 * 90) / 14, + ); + expect(airQualityChunkWeight("2015-01-01", "2015-01-01")).toBe(1); }); it("sends the coarse coordinates, the timezone and domains=auto, nothing else", async () => { @@ -268,6 +280,28 @@ describe("requests", () => { ]); }); + it("answers days before August 2022 outside Europe as uncoverable without a request", async () => { + const fetchSpy = vi.fn(async (_url: unknown) => + Response.json({ hourly: dayFixture("2022-08-01", 10) }), + ); + vi.stubGlobal("fetch", fetchSpy); + const result = await fetchDailyAirQuality({ + lat: 40.7, + lon: -74, + timezone: "UTC", + startDate: "2022-07-30", + endDate: "2022-08-01", + }); + expect(fetchSpy).toHaveBeenCalledTimes(1); + const url = new URL(String(fetchSpy.mock.calls[0]![0])); + expect(url.searchParams.get("start_date")).toBe("2022-08-01"); + expect( + result.days.filter((d) => d.aqHours === 0).map((d) => d.date), + ).toEqual(["2022-07-30", "2022-07-31"]); + expect(earliestAirQualityDay(40.7, -74)).toBe("2022-08-01"); + expect(earliestAirQualityDay(51.5, 7.2)).toBe("2013-01-01"); + }); + it("honours a self-hosted endpoint", async () => { vi.resetModules(); vi.stubEnv("OPENMETEO_AIR_QUALITY_URL", "https://aq.example.test/"); diff --git a/src/lib/environment/air-quality-history.ts b/src/lib/environment/air-quality-history.ts new file mode 100644 index 000000000..7d2ea80a2 --- /dev/null +++ b/src/lib/environment/air-quality-history.ts @@ -0,0 +1,589 @@ +/** + * v1.42 — the air-quality history backfill: air quality, pollen and UV for + * every past local day an account has entries on, not only from the day the + * module was switched on. + * + * ## Which days + * + * Every local day (in the zone the module keys its rows on: the home's, else + * the account's) with at least one measurement, mood entry, taken dose, + * symptom or workout, from the first day the feed serves up to eight days + * ago. The last week belongs to the nightly fetch, whose lookback already + * covers it (`ENVIRONMENT_LOOKBACK_DAYS`). A day is placed: + * + * - on its stored row's location when the day already has a weather row + * (only the air-quality part is fetched then, as the nightly gap fill + * does); + * - otherwise in the dated location period that covers it, else at the + * home, but the home only from `homeSince` on (or on every day when no + * `homeSince` is stored). A day before the home was set is never placed + * there: the weather and air of a place the person was not at would be + * wrong data in the correlations, the Coach and the day view, and an + * empty day is honest where a wrong one is not. Such a day is filled + * only when a period covers it. A day without a row gets its weather row + * with the air quality, because the air quality is stored on it. + * + * A day inside a period whose sealed location does not open is skipped (fail + * closed, as in the service), and so is a day with no place at all; neither + * counts towards the progress, which counts only days that can be filled. A day the + * source does not reach at its location (before 2013 in Europe, before August + * 2022 elsewhere, `earliestAirQualityDay`) is neither fetched nor counted: it + * stays without a value, which says nothing about the air, rather than + * carrying a zero. + * + * ## Requests + * + * The pending days are grouped by coarse location (the stored or resolved + * one, never finer: the same rounded coordinates the module always sends) + * and cut into ranges of at most 90 days, joining two days into one range + * when they lie at most 14 days apart. The feed weighs a request by days + * over 14 (`openMeteoCallWeight`), so a lone day costs as much as two weeks: + * joining short gaps is free, splitting long ones saves the gap. The newest + * ranges go first, so the window the correlations read fills first. The + * weather of a range is asked only over the span of its days without a row. + * + * ## Budget + * + * Every request still passes the instance budget and the account's daily + * share (`request-budget.ts`). On top of that the history holds itself to + * {@link AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS} of the account's 400 daily + * calls and to half of each instance window, so the nightly fetch, the gap + * fill and a backfill someone asked for keep their room. A run that reaches a + * ceiling stops; the nightly discovery picks the account up again, so a long + * history spreads over several days by construction. + * + * ## Idempotence and progress + * + * A day is done once its row carries `aqFetchedAt`; done days are never asked + * for again, and the writes only touch rows still without it (or create the + * row). Each run stores `{ total, done, checkedAt, complete }` on + * `User.environmentAqHistoryJson`, which the settings card shows as + * "Backfilling air quality: x of y days". + */ +import { Prisma } from "@/generated/prisma/client"; +import { prisma, toJson } from "@/lib/db"; +import { + airQualityChunkWeight, + earliestAirQualityDay, + fetchDailyAirQuality, + isAirQualityActive, + uncoverableDay, + AIR_QUALITY_CHUNK_DAYS, + AIR_QUALITY_EARLIEST_DAY, + type DailyAirQualityObservation, +} from "@/lib/environment/open-meteo-air-quality"; +import { + archiveRequestWeight, + fetchDailyEnvironment, + type DailyEnvironmentObservation, +} from "@/lib/environment/open-meteo"; +import { + OPEN_METEO_ACCOUNT_DAY_WINDOW, + OPEN_METEO_BUDGET_WINDOWS, + OpenMeteoBudgetExhaustedError, + readOpenMeteoBudgetUsage, + type BudgetUsage, +} from "@/lib/environment/request-budget"; +import { readLocation, sealLocation } from "@/lib/environment/location-cipher"; +import { + ENVIRONMENT_LOOKBACK_DAYS, + HOME_SELECT, + airQualityColumns, + buildUpsert, + inAnyPeriod, + readTravelOverrides, + resolveHome, + utcDayKey, + type HomeLocation, + type ResolvedLocation, + type TravelOverride, +} from "@/lib/environment/service"; +import { shiftDateKey } from "@/lib/tz/format"; +import { startOfLocalDayKey } from "@/lib/tz/local-day"; + +/** Days before today the history stops at; the nightly lookback has the rest. */ +export const AIR_QUALITY_HISTORY_SETTLE_DAYS = ENVIRONMENT_LOOKBACK_DAYS + 1; +/** Longest range one request covers (one feed chunk). */ +export const AIR_QUALITY_HISTORY_RANGE_DAYS = AIR_QUALITY_CHUNK_DAYS; +/** Two days at most this far apart share a range (14 days weigh one call). */ +export const AIR_QUALITY_HISTORY_JOIN_GAP_DAYS = 14; +/** Ranges one run works through before it hands over to its follow-up. */ +export const AIR_QUALITY_HISTORY_RANGES_PER_RUN = 8; +/** The history's ceiling inside the account's 400 daily calls. */ +export const AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS = Math.floor( + OPEN_METEO_ACCOUNT_DAY_WINDOW.limit * 0.75, +); +/** The history's share of each instance-wide window. */ +export const AIR_QUALITY_HISTORY_INSTANCE_SHARE = 0.5; + +/** One pending day and where it is fetched. */ +export interface HistoryDay { + date: string; + location: ResolvedLocation; + /** True when a weather row exists and only its air quality is missing. */ + hasRow: boolean; +} + +/** A contiguous request range over a sorted set of pending days. */ +export interface HistoryRange { + startDate: string; + endDate: string; + days: string[]; +} + +/** + * Cut sorted day keys into request ranges: a new range when the next day lies + * more than `joinGapDays` after the previous one, or when the range would + * grow past `maxRangeDays`. Pure; exported for tests. + */ +export function planHistoryRanges( + days: readonly string[], + opts: { joinGapDays?: number; maxRangeDays?: number } = {}, +): HistoryRange[] { + const joinGap = opts.joinGapDays ?? AIR_QUALITY_HISTORY_JOIN_GAP_DAYS; + const maxRange = opts.maxRangeDays ?? AIR_QUALITY_HISTORY_RANGE_DAYS; + const ranges: HistoryRange[] = []; + for (const day of [...new Set(days)].sort()) { + const current = ranges[ranges.length - 1]; + if ( + current && + shiftDateKey(current.endDate, joinGap) >= day && + shiftDateKey(current.startDate, maxRange - 1) >= day + ) { + current.endDate = day; + current.days.push(day); + } else { + ranges.push({ startDate: day, endDate: day, days: [day] }); + } + } + return ranges; +} + +/** + * The weight of one range in calls: its air-quality request, plus the + * weather request over the span of its days without a row, when it has any. + */ +export function historyRangeWeight( + range: HistoryRange, + missingRowDays: readonly string[], +): number { + const air = airQualityChunkWeight(range.startDate, range.endDate); + if (missingRowDays.length === 0) return air; + const sorted = [...missingRowDays].sort(); + return air + archiveRequestWeight(sorted[0], sorted[sorted.length - 1]); +} + +/** + * Whether a request of `weight` calls fits under the history's ceilings, given + * what each window already holds. Pure; exported for tests. + */ +export function historyRequestFits( + usage: BudgetUsage, + weight: number, +): boolean { + for (const w of OPEN_METEO_BUDGET_WINDOWS) { + const ceiling = w.limit * AIR_QUALITY_HISTORY_INSTANCE_SHARE; + if ((usage[w.name] ?? 0) + weight > ceiling) return false; + } + return ( + (usage["account-day"] ?? 0) + weight <= + AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS + ); +} + +/** + * Where a day without a row is placed: the period that covers it, else the + * home from `homeSince` on, or on every day when no `homeSince` is stored + * (see the header). Null for a day before the home with no period. Pure; + * exported for tests. + */ +export function resolveHistoryLocation( + day: string, + home: HomeLocation | null, + travels: readonly TravelOverride[], +): ResolvedLocation | null { + for (const t of travels) { + if (day >= t.startDate && day <= t.endDate) { + return { lat: t.lat, lon: t.lon, label: t.label, source: "TRAVEL" }; + } + } + if (home && (home.since === null || day >= home.since)) { + return { lat: home.lat, lon: home.lon, label: home.label, source: "HOME" }; + } + return null; +} + +/** The stored progress, as the settings card reads it. */ +export interface AirQualityHistoryState { + version: 1; + /** Past days with entries the source can cover at their place. */ + total: number; + /** Of those, the days whose air quality is stored. */ + done: number; + /** ISO instant of the run that counted them. */ + checkedAt: string; + /** True when nothing is left that the source can serve. */ + complete: boolean; +} + +/** Tolerant read of the stored progress; null for anything malformed. */ +export function readAirQualityHistoryState( + raw: unknown, +): AirQualityHistoryState | null { + if (raw === null || typeof raw !== "object") return null; + const r = raw as Record; + const count = (v: unknown) => + typeof v === "number" && Number.isInteger(v) && v >= 0 ? v : null; + const total = count(r.total); + const done = count(r.done); + if ( + total === null || + done === null || + typeof r.checkedAt !== "string" || + typeof r.complete !== "boolean" + ) { + return null; + } + return { + version: 1, + total, + done: Math.min(done, total), + checkedAt: r.checkedAt, + complete: r.complete, + }; +} + +/** + * The local days in `[fromKey, toKey]` with at least one entry: a reading, a + * mood entry, a taken dose, a symptom or a workout. Distinct in SQL, so a + * dense record costs a scan, not a transfer. `tz` is a zone the account + * stored (home or profile), bound as a parameter. + */ +export async function readEntryDays( + userId: string, + tz: string, + fromKey: string, + toKey: string, +): Promise { + // The instants of the first local day's start and the day after the last. + const from = startOfLocalDayKey(fromKey, tz); + const to = startOfLocalDayKey(shiftDateKey(toKey, 1), tz); + const rows = await prisma.$queryRaw>` + SELECT day FROM ( + SELECT DISTINCT to_char(("measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${tz}, 'YYYY-MM-DD') AS day + FROM "measurements" + WHERE "user_id" = ${userId} AND "deleted_at" IS NULL + AND "measured_at" >= ${from} AND "measured_at" < ${to} + UNION + SELECT "date" AS day FROM "mood_entries" + WHERE "user_id" = ${userId} AND "deleted_at" IS NULL + AND "date" >= ${fromKey} AND "date" <= ${toKey} + UNION + SELECT to_char(("taken_at" AT TIME ZONE 'UTC') AT TIME ZONE ${tz}, 'YYYY-MM-DD') AS day + FROM "medication_intake_events" + WHERE "user_id" = ${userId} AND "deleted_at" IS NULL + AND "taken_at" >= ${from} AND "taken_at" < ${to} + UNION + SELECT to_char(("occurred_at" AT TIME ZONE 'UTC') AT TIME ZONE ${tz}, 'YYYY-MM-DD') AS day + FROM "symptom_events" + WHERE "user_id" = ${userId} + AND "occurred_at" >= ${from} AND "occurred_at" < ${to} + UNION + SELECT to_char(("started_at" AT TIME ZONE 'UTC') AT TIME ZONE ${tz}, 'YYYY-MM-DD') AS day + FROM "workouts" + WHERE "user_id" = ${userId} + AND "started_at" >= ${from} AND "started_at" < ${to} + ) entry_days + `; + return rows + .map((r) => r.day) + .filter((d) => d >= fromKey && d <= toKey) + .sort(); +} + +/** What the history still has to do for an account, and how far it is. */ +export interface HistoryWork { + pending: HistoryDay[]; + total: number; + done: number; + timezone: string; +} + +/** + * Collect the pending days and the counts. Null when the account is gone or + * its air quality is off (for the account or the instance). + */ +export async function collectAirQualityHistory( + userId: string, + now: Date = new Date(), +): Promise { + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: HOME_SELECT, + }); + if (!user || !isAirQualityActive(user.environmentAirQualityEnabled)) { + return null; + } + const home = resolveHome(user); + const travel = await readTravelOverrides(userId); + const timezone = home?.timezone ?? user.timezone; + const cutoff = shiftDateKey(utcDayKey(now), -AIR_QUALITY_HISTORY_SETTLE_DAYS); + if (cutoff < AIR_QUALITY_EARLIEST_DAY) { + return { pending: [], total: 0, done: 0, timezone }; + } + + const [entryDays, rows] = await Promise.all([ + readEntryDays(userId, timezone, AIR_QUALITY_EARLIEST_DAY, cutoff), + prisma.environmentContext.findMany({ + where: { userId, date: { gte: AIR_QUALITY_EARLIEST_DAY, lte: cutoff } }, + select: { + date: true, + aqFetchedAt: true, + lat: true, + lon: true, + locationLabel: true, + locationEncrypted: true, + source: true, + }, + }), + ]); + const rowByDate = new Map(rows.map((r) => [r.date, r])); + + const pending: HistoryDay[] = []; + let total = 0; + let done = 0; + for (const date of entryDays) { + const row = rowByDate.get(date); + let location: ResolvedLocation | null; + if (row) { + const stored = readLocation({ + sealed: row.locationEncrypted, + lat: row.lat, + lon: row.lon, + label: row.locationLabel, + }); + location = stored ? { ...stored, source: row.source } : null; + } else if (inAnyPeriod(date, travel.unreadable)) { + location = null; + } else { + location = resolveHistoryLocation(date, home, travel.overrides); + } + if (!location) continue; + // Outside what the source serves at this place: left empty, not counted. + if (date < earliestAirQualityDay(location.lat, location.lon)) continue; + total += 1; + if (row?.aqFetchedAt) { + done += 1; + } else { + pending.push({ date, location, hasRow: row != null }); + } + } + return { pending, total, done, timezone }; +} + +/** A weather observation with nothing in it, for a day the archive skipped. */ +function emptyWeather(date: string): DailyEnvironmentObservation { + return { + date, + tempMin: null, + tempMax: null, + tempMean: null, + apparentMean: null, + apparentMax: null, + sunshineSec: null, + daylightSec: null, + precipSum: null, + pressureMean: null, + pressureDelta: null, + humidityMean: null, + cloudMean: null, + weatherCode: null, + }; +} + +export interface AirQualityHistoryRunResult { + /** `inactive`: the account is gone or its air quality is off. */ + status: "inactive" | "complete" | "progress" | "budget" | "error"; + /** Days whose air quality this run stored. */ + filled: number; + /** Requests sent. */ + fetches: number; + total: number; + done: number; + /** Days still pending after this run. */ + remaining: number; +} + +/** + * One run of the history for an account: up to `maxRanges` ranges, newest + * first, inside the history's budget ceilings. Stores the progress and says + * why it stopped. + */ +export async function runAirQualityHistory( + userId: string, + opts: { now?: Date; maxRanges?: number } = {}, +): Promise { + const now = opts.now ?? new Date(); + const maxRanges = opts.maxRanges ?? AIR_QUALITY_HISTORY_RANGES_PER_RUN; + const work = await collectAirQualityHistory(userId, now); + if (!work) { + return { + status: "inactive", + filled: 0, + fetches: 0, + total: 0, + done: 0, + remaining: 0, + }; + } + + // Group by the coarse coordinates the request carries, then plan ranges. + const groups = new Map< + string, + { lat: number; lon: number; byDate: Map } + >(); + for (const day of work.pending) { + const key = `${day.location.lat},${day.location.lon}`; + const group = groups.get(key) ?? { + lat: day.location.lat, + lon: day.location.lon, + byDate: new Map(), + }; + group.byDate.set(day.date, day); + groups.set(key, group); + } + const planned = [...groups.values()] + .flatMap((group) => + planHistoryRanges([...group.byDate.keys()]).map((range) => ({ + group, + range, + })), + ) + .sort((a, b) => b.range.endDate.localeCompare(a.range.endDate)); + + let filled = 0; + let fetches = 0; + let stop: "budget" | "error" | null = null; + let errors = 0; + let attempted = 0; + for (const { group, range } of planned) { + if (attempted >= maxRanges) break; + attempted += 1; + const days = range.days.map((d) => group.byDate.get(d)!); + const missing = days.filter((d) => !d.hasRow).map((d) => d.date); + const usage = await readOpenMeteoBudgetUsage(userId, new Date()); + if (!historyRequestFits(usage, historyRangeWeight(range, missing))) { + stop = "budget"; + break; + } + + const air = await fetchDailyAirQuality( + { + lat: group.lat, + lon: group.lon, + timezone: work.timezone, + startDate: range.startDate, + endDate: range.endDate, + }, + { accountId: userId }, + ); + if (air.stopped === "budget") { + stop = "budget"; + break; + } + if (air.stopped === "error") { + errors += 1; + continue; + } + fetches += 1; + const airByDay = new Map( + air.days.map((d) => [d.date, d]), + ); + const airFor = (date: string) => airByDay.get(date) ?? uncoverableDay(date); + + let weather: Map | null = null; + if (missing.length > 0) { + try { + const observations = await fetchDailyEnvironment( + { + lat: group.lat, + lon: group.lon, + timezone: work.timezone, + startDate: missing[0], + endDate: missing[missing.length - 1], + }, + { accountId: userId }, + ); + fetches += 1; + weather = new Map(observations.map((o) => [o.date, o])); + } catch (err) { + if (err instanceof OpenMeteoBudgetExhaustedError) { + stop = "budget"; + } else { + errors += 1; + } + // The days with a row are stored below; those without one stay + // pending until the weather can be fetched with them. + } + } + + const fetchedAt = new Date(); + const ops: Prisma.PrismaPromise[] = []; + for (const day of days) { + if (day.hasRow) { + ops.push( + prisma.environmentContext.updateMany({ + // Still unfetched, so a weather rewrite that landed meanwhile + // (with its own air quality) is not overwritten. + where: { userId, date: day.date, aqFetchedAt: null }, + data: airQualityColumns(airFor(day.date), fetchedAt), + }), + ); + } else if (weather) { + ops.push( + buildUpsert( + userId, + day.location, + sealLocation(day.location), + weather.get(day.date) ?? emptyWeather(day.date), + { active: true, day: airFor(day.date), now: fetchedAt }, + ), + ); + } + } + if (ops.length > 0) { + const results = await prisma.$transaction(ops); + for (const r of results) { + filled += + r && typeof r === "object" && "count" in r + ? Number((r as { count: number }).count) + : 1; + } + } + if (stop) break; + } + + const done = Math.min(work.total, work.done + filled); + const remaining = work.total - done; + const state: AirQualityHistoryState = { + version: 1, + total: work.total, + done, + checkedAt: now.toISOString(), + complete: remaining === 0, + }; + await prisma.user.update({ + where: { id: userId }, + data: { environmentAqHistoryJson: toJson(state) }, + }); + + const status: AirQualityHistoryRunResult["status"] = + remaining === 0 + ? "complete" + : stop === "budget" + ? "budget" + : errors > 0 && filled === 0 + ? "error" + : "progress"; + return { status, filled, fetches, total: work.total, done, remaining }; +} diff --git a/src/lib/environment/open-meteo-air-quality.ts b/src/lib/environment/open-meteo-air-quality.ts index 4d2c8a917..85d44b373 100644 --- a/src/lib/environment/open-meteo-air-quality.ts +++ b/src/lib/environment/open-meteo-air-quality.ts @@ -22,14 +22,20 @@ * * ## Coverage, and why a missing value is null and never zero * - * The feed starts on 2013-01-01; pollen, UV, dust and aerosol depth only from - * about mid 2022, and pollen only in Europe. A value the feed did not cover is - * stored as null. A chunk that ends before {@link AIR_QUALITY_FULL_FEED_FROM} - * asks only for the eight pollutant and index variables, which the feed has - * for that era; that keeps the request weight at 1.0 instead of 1.7 per ten - * variables. A day before {@link AIR_QUALITY_EARLIEST_DAY} cannot be fetched - * at all and is returned as uncoverable (every value null, no hours), so the - * nightly gap fill does not ask for it again. + * What the feed serves (open-meteo.com/en/docs/air-quality-api, checked + * against the live API in v1.42): inside Europe the CAMS European reanalysis + * reaches back to 2013-01-01 and the forecast takes over from October 2023; + * outside Europe only CAMS global exists, from 2022-08-01. Pollen is Europe + * only and present from 2021 (in season); UV and aerosol depth come from CAMS + * global, so from 2022-08-01. A value the feed did not cover is stored as + * null. The variables asked for follow the era of a chunk's last day + * ({@link variablesForChunk}): the eight pollutant and index variables before + * {@link AIR_QUALITY_POLLEN_FROM}, plus pollen and dust until + * {@link AIR_QUALITY_GLOBAL_FROM}, everything after, which keeps the + * request weight down where the extra variables would come back empty. A day + * before the first day the feed serves at the location + * ({@link earliestAirQualityDay}) is not requested at all and is returned as + * uncoverable (every value null, no hours), so nobody asks for it again. * * ## Daily rules (a local day of hourly values) * @@ -90,13 +96,15 @@ export function isAirQualityActive(accountEnabled: boolean): boolean { return accountEnabled && !isAirQualityOperatorDisabled(); } -/** The first day the feed serves. */ +/** The first day the feed serves (the CAMS European reanalysis). */ export const AIR_QUALITY_EARLIEST_DAY = "2013-01-01"; /** - * From about here the feed carries pollen, UV, dust and aerosol depth. A - * chunk ending before it asks for the pollutant and index variables only. + * The first day CAMS global serves, the only model outside Europe, and so the + * first day with every variable (UV and aerosol depth come from it). */ -export const AIR_QUALITY_FULL_FEED_FROM = "2022-06-01"; +export const AIR_QUALITY_GLOBAL_FROM = "2022-08-01"; +/** From here the European feed carries pollen (and dust). */ +export const AIR_QUALITY_POLLEN_FROM = "2021-01-01"; /** Days per request: bounds the response (about 300 KB) and the heap. */ export const AIR_QUALITY_CHUNK_DAYS = 90; @@ -112,11 +120,9 @@ const POLLUTANT_VARIABLES = [ "us_aqi", ] as const; -/** The nine variables the feed fills only from about mid 2022. */ -const RECENT_VARIABLES = [ - "uv_index", +/** The seven variables the European feed fills from 2021. */ +const POLLEN_ERA_VARIABLES = [ "dust", - "aerosol_optical_depth", "alder_pollen", "birch_pollen", "grass_pollen", @@ -125,13 +131,23 @@ const RECENT_VARIABLES = [ "ragweed_pollen", ] as const; +/** The two variables that come from CAMS global, so from August 2022. */ +const GLOBAL_ERA_VARIABLES = ["uv_index", "aerosol_optical_depth"] as const; + type HourlyVariable = - (typeof POLLUTANT_VARIABLES)[number] | (typeof RECENT_VARIABLES)[number]; + | (typeof POLLUTANT_VARIABLES)[number] + | (typeof POLLEN_ERA_VARIABLES)[number] + | (typeof GLOBAL_ERA_VARIABLES)[number]; + +const POLLEN_ERA_REQUEST: readonly HourlyVariable[] = [ + ...POLLUTANT_VARIABLES, + ...POLLEN_ERA_VARIABLES, +]; /** All 17 hourly variables, in request order. */ export const AIR_QUALITY_VARIABLES: readonly HourlyVariable[] = [ - ...POLLUTANT_VARIABLES, - ...RECENT_VARIABLES, + ...POLLEN_ERA_REQUEST, + ...GLOBAL_ERA_VARIABLES, ]; /** One stored day's air-quality part, as the feed produced it. */ @@ -211,6 +227,16 @@ export function insideCamsEurope(lat: number, lon: number): boolean { return lat >= 30 && lat <= 72 && lon >= -25 && lon <= 45; } +/** + * The first day the feed has air quality for at a coarse location: 2013 + * inside the CAMS Europe domain, August 2022 (CAMS global) outside it. + */ +export function earliestAirQualityDay(lat: number, lon: number): string { + return insideCamsEurope(lat, lon) + ? AIR_QUALITY_EARLIEST_DAY + : AIR_QUALITY_GLOBAL_FROM; +} + /** * Fold an hourly response into one observation per local day present in it. * Pure; exported for tests. Times are the feed's local `YYYY-MM-DDTHH:MM`; @@ -326,11 +352,22 @@ export function chunkDays( return chunks; } -/** The variables a chunk asks for: the pollutant set alone before the full era. */ +/** The variables a chunk asks for, by the era of its last day. */ export function variablesForChunk(endDate: string): readonly HourlyVariable[] { - return endDate < AIR_QUALITY_FULL_FEED_FROM - ? POLLUTANT_VARIABLES - : AIR_QUALITY_VARIABLES; + if (endDate < AIR_QUALITY_POLLEN_FROM) return POLLUTANT_VARIABLES; + if (endDate < AIR_QUALITY_GLOBAL_FROM) return POLLEN_ERA_REQUEST; + return AIR_QUALITY_VARIABLES; +} + +/** The request weight of one chunk, in calls (`openMeteoCallWeight`). */ +export function airQualityChunkWeight( + startDate: string, + endDate: string, +): number { + return openMeteoCallWeight( + variablesForChunk(endDate).length, + enumerateDayCount(startDate, endDate), + ); } /** Why a fetch stopped before its last chunk. */ @@ -361,27 +398,23 @@ export async function fetchDailyAirQuality( budget: { accountId?: string } = {}, ): Promise { const days: DailyAirQualityObservation[] = []; - // Days before the feed's first day are answered here, without a request. + // Days before the feed's first day at this location are answered here, + // without a request. + const earliest = earliestAirQualityDay(args.lat, args.lon); for ( let day = args.startDate; - day <= args.endDate && day < AIR_QUALITY_EARLIEST_DAY; + day <= args.endDate && day < earliest; day = shiftDateKey(day, 1) ) { days.push(uncoverableDay(day)); } - const start = - args.startDate < AIR_QUALITY_EARLIEST_DAY - ? AIR_QUALITY_EARLIEST_DAY - : args.startDate; + const start = args.startDate < earliest ? earliest : args.startDate; if (start > args.endDate) return { days, stopped: null }; const chunks = chunkDays(start, args.endDate).reverse(); for (const chunk of chunks) { const variables = variablesForChunk(chunk.endDate); - const weight = openMeteoCallWeight( - variables.length, - enumerateDayCount(chunk.startDate, chunk.endDate), - ); + const weight = airQualityChunkWeight(chunk.startDate, chunk.endDate); if (!(await reserveOpenMeteoCalls(weight, budget.accountId))) { return { days, stopped: "budget" }; } diff --git a/src/lib/environment/open-meteo.ts b/src/lib/environment/open-meteo.ts index f936d5c4a..ed3245e14 100644 --- a/src/lib/environment/open-meteo.ts +++ b/src/lib/environment/open-meteo.ts @@ -149,6 +149,17 @@ const HOURLY_FIELDS = HOURLY_FIELD_LIST.join(","); const ARCHIVE_VARIABLE_COUNT = DAILY_FIELD_LIST.length + HOURLY_FIELD_LIST.length; +/** The request weight of one archive request over a day range, in calls. */ +export function archiveRequestWeight( + startDate: string, + endDate: string, +): number { + return openMeteoCallWeight( + ARCHIVE_VARIABLE_COUNT, + enumerateDayCount(startDate, endDate), + ); +} + /** Per-day aggregation accumulator for the hourly-only fields. */ interface HourlyDayAgg { pressures: number[]; @@ -269,10 +280,7 @@ export async function fetchDailyEnvironment( }, budget: { accountId?: string } = {}, ): Promise { - const weight = openMeteoCallWeight( - ARCHIVE_VARIABLE_COUNT, - enumerateDayCount(args.startDate, args.endDate), - ); + const weight = archiveRequestWeight(args.startDate, args.endDate); if (!(await reserveOpenMeteoCalls(weight, budget.accountId))) { throw new OpenMeteoBudgetExhaustedError(); } diff --git a/src/lib/environment/request-budget.ts b/src/lib/environment/request-budget.ts index 58f0c31c5..05f6d46fa 100644 --- a/src/lib/environment/request-budget.ts +++ b/src/lib/environment/request-budget.ts @@ -171,3 +171,39 @@ export class OpenMeteoBudgetExhaustedError extends Error { this.name = "OpenMeteoBudgetExhaustedError"; } } + +/** Calls already charged to each live window, by window name. */ +export type BudgetUsage = Partial>; + +/** + * Read how much of each window is in use, in calls, without charging + * anything. A window never written or already reset reads as absent (zero). + * With `accountId` the account's daily share is read as well. Used by the + * air-quality history backfill, which holds itself under a ceiling of its own + * so the nightly fetch and on-demand requests keep their room. + */ +export async function readOpenMeteoBudgetUsage( + accountId?: string, + now: Date = new Date(), +): Promise { + const keys = new Map( + OPEN_METEO_BUDGET_WINDOWS.map((w) => [ + `${BUCKET_PREFIX}:${w.name}`, + w.name, + ]), + ); + if (accountId) { + keys.set(`${BUCKET_PREFIX}:account:${accountId}:day`, "account-day"); + } + const rows = await prisma.rateLimit.findMany({ + where: { key: { in: [...keys.keys()] } }, + select: { key: true, count: true, resetAt: true }, + }); + const usage: BudgetUsage = {}; + for (const row of rows) { + const name = keys.get(row.key); + if (!name || row.resetAt.getTime() <= now.getTime()) continue; + usage[name] = row.count / CENTI; + } + return usage; +} diff --git a/src/lib/environment/service.ts b/src/lib/environment/service.ts index bd26c43f0..d9af4cea7 100644 --- a/src/lib/environment/service.ts +++ b/src/lib/environment/service.ts @@ -56,14 +56,14 @@ export const ENVIRONMENT_LOOKBACK_DAYS = 7; export const ENVIRONMENT_MAX_BACKFILL_DAYS = 730; /** A coarse resolved location for a given day. */ -interface ResolvedLocation { +export interface ResolvedLocation { lat: number; lon: number; label: string; source: EnvironmentLocationSource; } -interface HomeLocation { +export interface HomeLocation { lat: number; lon: number; label: string; @@ -72,7 +72,7 @@ interface HomeLocation { since: string | null; } -interface TravelOverride { +export interface TravelOverride { startDate: string; endDate: string; lat: number; @@ -164,7 +164,7 @@ export interface FetchAndStoreResult { } /** The user columns every environment read resolves the home from. */ -const HOME_SELECT = { +export const HOME_SELECT = { homeLat: true, homeLon: true, homeLabel: true, @@ -208,7 +208,7 @@ export function resolveHome(user: { } /** A location period whose sealed location this host cannot open. */ -interface UnreadablePeriod { +export interface UnreadablePeriod { startDate: string; endDate: string; } @@ -220,7 +220,7 @@ interface UnreadablePeriod { * period it cannot read still says "this account was not at home on these * days", so those days must not fall through to the home location. */ -async function readTravelOverrides(userId: string): Promise<{ +export async function readTravelOverrides(userId: string): Promise<{ overrides: TravelOverride[]; unreadable: UnreadablePeriod[]; }> { @@ -260,7 +260,7 @@ async function readTravelOverrides(userId: string): Promise<{ } /** True when `day` falls inside one of `periods`. */ -function inAnyPeriod(day: string, periods: readonly UnreadablePeriod[]) { +export function inAnyPeriod(day: string, periods: readonly UnreadablePeriod[]) { return periods.some((p) => day >= p.startDate && day <= p.endDate); } @@ -401,7 +401,10 @@ export async function fetchAndStoreEnvironment(args: { } /** The air-quality columns of a stored day, from one observation. */ -function airQualityColumns(day: DailyAirQualityObservation, fetchedAt: Date) { +export function airQualityColumns( + day: DailyAirQualityObservation, + fetchedAt: Date, +) { return { pm25Mean: day.pm25Mean, pm25Max: day.pm25Max, @@ -453,7 +456,7 @@ const AIR_QUALITY_UNFETCHED = { } as const; /** Build the idempotent per-day upsert operation (un-awaited, for a batch). */ -function buildUpsert( +export function buildUpsert( userId: string, loc: ResolvedLocation, sealed: Uint8Array, @@ -523,7 +526,8 @@ export interface AirQualityGapFillResult { * null), newest first, so the correlation window fills first and an account * with two years of weather catches up within about nine nights. Each day is * fetched at the location it was stored for (opened from its sealed copy), - * never re-resolved. A day the feed cannot serve (before 2013) is marked + * never re-resolved. A day the feed cannot serve (before 2013 in Europe, + * before August 2022 elsewhere) is marked * fetched with every value null, so it is not asked for again. A no-op when * the air-quality part is off for the account or the instance. */ diff --git a/src/lib/export/backup-plan.ts b/src/lib/export/backup-plan.ts index 9a07f0bce..e33dfac32 100644 --- a/src/lib/export/backup-plan.ts +++ b/src/lib/export/backup-plan.ts @@ -909,6 +909,9 @@ export const USER_COLUMN_BACKUP_CLASS = { // The restore stamps its own: paired clients' cursors from before it must // be refused, which a carried value would undo. syncResetAt: "OPERATIONAL", + // v1.42 — the air-quality history backfill's progress; the restored + // account's job recounts it on its next run. + environmentAqHistoryJson: "OPERATIONAL", // ── Settings ────────────────────────────────────────────────────────────── // The body and the person, for the reference ranges, the score and the diff --git a/src/lib/jobs/environment-air-quality-history.ts b/src/lib/jobs/environment-air-quality-history.ts new file mode 100644 index 000000000..f63423b19 --- /dev/null +++ b/src/lib/jobs/environment-air-quality-history.ts @@ -0,0 +1,246 @@ +/** + * v1.42 — the air-quality history backfill queue + * (`src/lib/environment/air-quality-history.ts` does the work). + * + * Three ways in, all coalesced per account by `singletonKey`: + * - the boot discovery after an update and the nightly discovery tick + * (02:40 Europe/Berlin, after the 02:10 environment fetch), which offer + * every account with the module and its air quality on, a place to + * resolve days to, and a history that is not through or was last checked + * more than a week ago (old entries imported since); + * - setting the home location; + * - switching the air-quality part on. + * + * One run works through a bounded number of ranges and stores its progress. + * When it made progress and the history is not through, and no budget + * ceiling stopped it, it sends its own follow-up half a minute later. A run + * the budget stopped sends none: the next nightly discovery resumes it, which + * is what spreads a long history over several days. The module gate and the + * air-quality switches are re-read on every run, so switching either off + * ends the chain. + * + * The queue name MUST be registered in `allQueues` in + * `src/lib/jobs/reminder/register-maintenance.ts`, or the sends silently + * no-op. + */ +import type { PgBoss } from "pg-boss"; + +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { + getOperatorModuleAvailability, + isModuleEnabled, + normalisePrefs, + resolveModuleEnabled, +} from "@/lib/modules/gate"; +import { isAirQualityOperatorDisabled } from "@/lib/environment/open-meteo-air-quality"; +import { + readAirQualityHistoryState, + runAirQualityHistory, +} from "@/lib/environment/air-quality-history"; +import { workerLog } from "./reminder/shared"; + +export const ENVIRONMENT_AQ_HISTORY_QUEUE = "environment-aq-history"; + +/** Daily 02:40 Europe/Berlin, after the nightly environment fetch. */ +export const ENVIRONMENT_AQ_HISTORY_CRON = "40 2 * * *"; + +/** Delay before a run's follow-up, in seconds. */ +export const AQ_HISTORY_CONTINUATION_DELAY_SECONDS = 30; + +/** Follow-ups one chain may send before it waits for the next discovery. */ +export const AQ_HISTORY_MAX_CONTINUATIONS = 50; + +/** A history checked longer ago than this is offered again (new old entries). */ +const AQ_HISTORY_RECHECK_MS = 7 * 24 * 60 * 60 * 1000; + +export interface EnvironmentAqHistoryPayload { + /** Absent on the discovery tick. */ + userId?: string; + continuation?: number; +} + +function singletonKey(userId: string): string { + return `environment-aq-history:${userId}`; +} + +/** + * Queue a history run for one account. No-ops cleanly without a worker (the + * nightly discovery still covers it). A send while a run is queued collapses + * into it. + */ +export async function enqueueAirQualityHistory( + userId: string, + startAfterSeconds: number = 0, +): Promise { + const boss = getGlobalBoss(); + if (!boss) return false; + const id = await boss.send( + ENVIRONMENT_AQ_HISTORY_QUEUE, + { userId } satisfies EnvironmentAqHistoryPayload, + { + singletonKey: singletonKey(userId), + ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), + }, + ); + return id != null; +} + +/** Whether a stored progress means the account should be offered a run. */ +export function historyDue(raw: unknown, now: Date = new Date()): boolean { + const state = readAirQualityHistoryState(raw); + if (!state || !state.complete) return true; + const checked = Date.parse(state.checkedAt); + return ( + !Number.isFinite(checked) || now.getTime() - checked > AQ_HISTORY_RECHECK_MS + ); +} + +/** + * Discovery: one run per account with the module on, its air quality on, a + * home or a location period, and a history that is due. Nothing at all when + * the operator turned air quality off. + */ +export async function discoverAirQualityHistory( + boss: PgBoss, + startAfterSeconds: number = 0, +): Promise<{ enqueued: number; skipped: number }> { + if (isAirQualityOperatorDisabled()) return { enqueued: 0, skipped: 0 }; + const [candidates, operatorAvailability] = await Promise.all([ + prisma.user.findMany({ + where: { + environmentAirQualityEnabled: true, + OR: [ + { homeLocationEncrypted: { not: null } }, + { homeLat: { not: null }, homeLon: { not: null } }, + { environmentTravelLocations: { some: {} } }, + ], + }, + select: { + id: true, + modulePreferencesJson: true, + environmentAqHistoryJson: true, + }, + }), + getOperatorModuleAvailability(), + ]); + + const now = new Date(); + let enqueued = 0; + let skipped = 0; + for (const candidate of candidates) { + const moduleOn = resolveModuleEnabled( + "environment", + { + gender: null, + disableCoach: false, + modulePreferences: normalisePrefs(candidate.modulePreferencesJson), + cycleTrackingEnabled: null, + }, + false, + operatorAvailability, + ); + if (!moduleOn || !historyDue(candidate.environmentAqHistoryJson, now)) { + skipped += 1; + continue; + } + await boss.send( + ENVIRONMENT_AQ_HISTORY_QUEUE, + { userId: candidate.id } satisfies EnvironmentAqHistoryPayload, + { + singletonKey: singletonKey(candidate.id), + ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), + }, + ); + enqueued += 1; + } + return { enqueued, skipped }; +} + +/** + * Boot discovery, staggered past the startup storm. Best-effort: the error + * comes back in the result so worker boot never fails on it. + */ +export async function enqueueBootTimeAirQualityHistory( + startAfterSeconds: number, +): Promise<{ enqueued: number; skipped: number; error: string | null }> { + const boss = getGlobalBoss(); + if (!boss) return { enqueued: 0, skipped: 0, error: null }; + try { + return { + ...(await discoverAirQualityHistory(boss, startAfterSeconds)), + error: null, + }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} + +/** The queue handler: discovery for an empty payload, else one account's run. */ +export async function handleEnvironmentAqHistory( + boss: PgBoss, + payload: EnvironmentAqHistoryPayload, +): Promise { + if (!payload.userId) { + const { enqueued, skipped } = await discoverAirQualityHistory(boss); + workerLog( + "info", + `[environment-aq-history] discovery enqueued=${enqueued} skipped=${skipped}`, + ); + return jobDone({ + discovery_enqueued: enqueued, + discovery_skipped: skipped, + }); + } + + const userId = payload.userId; + if (!(await isModuleEnabled(userId, "environment"))) { + return jobDone({ skipped: "environment_module_disabled" }); + } + + const result = await runAirQualityHistory(userId); + const continuation = payload.continuation ?? 0; + let continued = false; + if ( + result.status === "progress" && + result.filled > 0 && + continuation < AQ_HISTORY_MAX_CONTINUATIONS + ) { + await boss.send( + ENVIRONMENT_AQ_HISTORY_QUEUE, + { + userId, + continuation: continuation + 1, + } satisfies EnvironmentAqHistoryPayload, + { + singletonKey: singletonKey(userId), + startAfter: AQ_HISTORY_CONTINUATION_DELAY_SECONDS, + }, + ); + continued = true; + } + workerLog( + "info", + `[environment-aq-history] user=${userId} status=${result.status} filled=${result.filled} done=${result.done}/${result.total} fetches=${result.fetches}${continued ? " continued" : ""}`, + ); + // Every range failed at the feed: say so, so the failing-jobs card sees a + // feed that keeps refusing. A budget stop is not a failure: nothing was + // sent past the ceiling, and the next discovery resumes where it stopped. + if (result.status === "error") { + return jobFailed("air-quality history: every range failed", undefined, { + fetches: result.fetches, + }); + } + return jobDone({ + outcome: result.status, + days_stored: result.filled, + fetches: result.fetches, + budget_blocked: result.status === "budget", + continued, + }); +} diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 9d3e564c6..9b610fd8a 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -466,4 +466,7 @@ export const QUEUE_RUNTIME: Readonly> = { "environment-fetch": short( "A fan-out of sends, or one account's bounded archive range.", ), + "environment-aq-history": short( + "A fan-out of sends, or one account's history run: at most eight ranges of up to 90 days, two requests each, then a follow-up.", + ), }; diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 801a38ec1..0458351f0 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -197,6 +197,13 @@ import { handleEnvironmentFetch, type EnvironmentFetchPayload, } from "@/lib/jobs/environment-fetch"; +import { + ENVIRONMENT_AQ_HISTORY_CRON, + ENVIRONMENT_AQ_HISTORY_QUEUE, + enqueueBootTimeAirQualityHistory, + handleEnvironmentAqHistory, + type EnvironmentAqHistoryPayload, +} from "@/lib/jobs/environment-air-quality-history"; import { recordError } from "@/lib/jobs/worker-status"; import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { @@ -419,6 +426,9 @@ const allQueues = [ // surface. Without this entry pg-boss never provisions it and both the cron // and the backfill button silently no-op. ENVIRONMENT_FETCH_QUEUE, + // v1.42 — the air-quality history backfill: per account, from boot and + // nightly discovery, the home being set and air quality switched on. + ENVIRONMENT_AQ_HISTORY_QUEUE, DATA_BACKUP_QUEUE, RATE_LIMIT_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_QUEUE, @@ -600,6 +610,9 @@ const schedules: ScheduleEntry[] = [ // v1.25 (W-ENV) — daily 02:10 Europe/Berlin discovery tick (empty payload) // that fans out one per-user environment fetch per opted-in account. [ENVIRONMENT_FETCH_QUEUE, ENVIRONMENT_FETCH_CRON], + // v1.42 — daily 02:40 discovery for the air-quality history backfill, after + // the night's environment fetch has taken its share of the budget. + [ENVIRONMENT_AQ_HISTORY_QUEUE, ENVIRONMENT_AQ_HISTORY_CRON], // The send options are the point of this tuple: without them the weekly // snapshot inherits pg-boss's 15-minute expiration and is killed mid-run on // any sizeable record. See `data-backup-policy.ts` for why the window rides @@ -823,6 +836,15 @@ const queuePolicies: QueuePolicyTable = { // active. A boot's send for a queued account collapses into it; a duplicate // beside an active run rewrites nothing, since only differing means are // written. + // v1.42 — the air-quality history backfill. `short` for the same reason: a + // run with work left sends its own follow-up while it is still active. A + // send for an account whose run is queued collapses into it; one beside an + // active run only re-reads what is done, since done days are never fetched. + [ENVIRONMENT_AQ_HISTORY_QUEUE]: { + policy: "short", + reason: + "Per-account singleton key; collapse queued duplicates only, so the active run can queue its own follow-up.", + }, [MEASUREMENT_FOLD_REPAIR_QUEUE]: { policy: "short", reason: @@ -1769,6 +1791,22 @@ export async function registerMaintenanceQueues( }, ); + // v1.42 — the air-quality history backfill. Serial: every run draws on the + // same instance-wide Open-Meteo budget as the nightly fetch. + await createAndWork( + boss, + ENVIRONMENT_AQ_HISTORY_QUEUE, + { localConcurrency: 1 }, + async (jobs) => { + let failure: JobOutcome | null = null; + for (const job of jobs) { + const outcome = await handleEnvironmentAqHistory(boss, job.data); + if (!outcome.ok && failure === null) failure = outcome; + } + return failure ?? jobDone({ jobs: jobs.length }); + }, + ); + // v1.25 (W-ENV) — nightly environment fetch. The daily discovery tick (empty // payload) fans out one per-user job per opted-in account; the queue also // serves the on-demand backfill payloads from the settings surface. Serial @@ -1925,6 +1963,31 @@ export async function enqueueMaintenanceBootDiscovery(): Promise { ); } + // v1.42 — the air-quality history backfill after an update, staggered past + // the boot storm. Accounts whose history is through are not offered. + try { + const { enqueued, skipped, error } = await enqueueBootTimeAirQualityHistory( + BOOT_BACKFILL_STAGGER_SECONDS * 7, + ); + if (error) { + workerLog( + "error", + `[environment-aq-history] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[environment-aq-history] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[environment-aq-history] boot discovery threw an unexpected error", + err, + ); + } + // v1.42 — the fold repair, once per account; the purge below waits for it. try { const { enqueued, skipped, error } = diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index 2d144f23d..deda17e04 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -94,6 +94,7 @@ const subsystemSurface = { // register-maintenance.ts "environment-fetch": { audience: "account" }, + "environment-aq-history": { audience: "system" }, "data-backup": { audience: "system" }, "rate-limit-cleanup": { audience: "system" }, "idempotency-cleanup": { audience: "system" }, diff --git a/src/lib/openapi/routes/documents.ts b/src/lib/openapi/routes/documents.ts index ce08c12f3..419876ce4 100644 --- a/src/lib/openapi/routes/documents.ts +++ b/src/lib/openapi/routes/documents.ts @@ -62,7 +62,7 @@ const documentsLayoutSchema = z arrangement: z .enum(DOCUMENTS_LAYOUT_ARRANGEMENTS) .describe( - '`stacked` gives every month its own block; `flow` runs the documents across the full width and wraps, with month names inline. Default "stacked".', + '`stacked` gives every month its own block; `flow` runs the documents across the full width and wraps, with month names inline. Default "flow"; an account that never chose keeps following the default.', ), }) .meta({ diff --git a/src/lib/openapi/routes/environment.ts b/src/lib/openapi/routes/environment.ts index c14803c29..ca4ffcc6b 100644 --- a/src/lib/openapi/routes/environment.ts +++ b/src/lib/openapi/routes/environment.ts @@ -163,6 +163,29 @@ const environmentOverview = z .describe( "The model domain of that day: `cams_europe` (about 11 km, with pollen) or `cams_global` (about 45 km, no pollen).", ), + history: z + .object({ + total: z + .number() + .int() + .describe( + "Past days with entries the source can serve at their place (from 2013 in Europe, from August 2022 elsewhere). Older days stay without values.", + ), + done: z + .number() + .int() + .describe("Of those, the days whose air quality is stored."), + complete: z + .boolean() + .describe("True when nothing the source serves is left."), + checkedAt: z + .string() + .describe("ISO instant of the run that counted them."), + }) + .nullable() + .describe( + "Progress of the background backfill that fills air quality for every past day with entries. Null before its first run and while air quality is off.", + ), }) .describe("v1.42 — the air-quality part of the module."), latestDay: z diff --git a/tests/integration/environment-air-quality-history.test.ts b/tests/integration/environment-air-quality-history.test.ts new file mode 100644 index 000000000..b71bf55ae --- /dev/null +++ b/tests/integration/environment-air-quality-history.test.ts @@ -0,0 +1,368 @@ +/** + * v1.42 — the air-quality history backfill against real Postgres, with + * Open-Meteo mocked at the wire. + * + * - every past local day with an entry is filled: a day with a weather row + * gets its air quality, a day without one gets both, a day of a trip is + * fetched at the trip's place, a day from the home's effective date at + * the home; + * - a day before the home was set is filled only inside a location period; + * without one it stays empty and is not counted; + * - a day before the source reaches stays without a row and is not counted, + * and the last week is left to the nightly fetch; + * - nearby days share one request, the coordinates sent are the coarse + * ones, and a second run sends nothing (idempotent); + * - the history's own ceiling inside the account's daily share stops a run + * before anything is sent, and the progress says how far it got; + * - the account switch and the operator switch stop it entirely; + * - the queue handler runs it and the overview reports the progress. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS, + runAirQualityHistory, +} from "@/lib/environment/air-quality-history"; +import { openLocation, sealLocation } from "@/lib/environment/location-cipher"; +import { handleEnvironmentAqHistory } from "@/lib/jobs/environment-air-quality-history"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const OWNER = "aq-history-owner"; +const NOW = new Date("2026-10-09T03:00:00Z"); +const BOCHUM = { lat: 51.5, lon: 7.2, label: "Bochum" }; +const LISBON = { lat: 38.7, lon: -9.1, label: "Lisbon" }; +const ESSEN = { lat: 51.5, lon: 7.0, label: "Essen" }; + +function daysBetween(start: string, end: string): string[] { + const out: string[] = []; + for ( + let t = Date.parse(`${start}T00:00:00Z`); + t <= Date.parse(`${end}T00:00:00Z`); + t += 86_400_000 + ) { + out.push(new Date(t).toISOString().slice(0, 10)); + } + return out; +} + +/** Answers both feeds with a full day of values for every requested day. */ +function openMeteoStub() { + return vi.fn(async (input: unknown) => { + const url = new URL(String(input)); + const days = daysBetween( + url.searchParams.get("start_date")!, + url.searchParams.get("end_date")!, + ); + if (url.pathname === "/v1/air-quality") { + const variables = url.searchParams.get("hourly")!.split(","); + const time = days.flatMap((d) => + Array.from( + { length: 24 }, + (_, h) => `${d}T${String(h).padStart(2, "0")}:00`, + ), + ); + const hourly: Record = { time }; + for (const v of variables) hourly[v] = time.map(() => 12); + return Response.json({ hourly }); + } + const daily: Record = { time: days }; + for (const field of url.searchParams.get("daily")!.split(",")) { + daily[field] = days.map(() => 5); + } + const time = days.flatMap((d) => + Array.from( + { length: 24 }, + (_, h) => `${d}T${String(h).padStart(2, "0")}:00`, + ), + ); + const hourly: Record = { time }; + for (const field of url.searchParams.get("hourly")!.split(",")) { + hourly[field] = time.map(() => 1000); + } + return Response.json({ daily, hourly }); + }); +} + +async function measurementOn(day: string) { + await getPrismaClient().measurement.create({ + data: { + userId: OWNER, + type: "WEIGHT", + value: 70, + unit: "kg", + measuredAt: new Date(`${day}T10:00:00Z`), + }, + }); +} + +let fetchSpy: ReturnType; + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: OWNER, + username: OWNER, + email: `${OWNER}@example.test`, + timezone: "Europe/Berlin", + homeLocationEncrypted: sealLocation(BOCHUM), + homeTimezone: "Europe/Berlin", + // Days before it are filled only inside a location period. + homeSince: new Date("2025-01-01T00:00:00Z"), + modulePreferencesJson: { environment: true }, + }, + }); + await prisma.environmentTravelLocation.create({ + data: { + userId: OWNER, + startDate: "2024-07-01", + endDate: "2024-07-14", + locationEncrypted: sealLocation(LISBON), + }, + }); + await prisma.environmentTravelLocation.create({ + data: { + userId: OWNER, + startDate: "2020-03-01", + endDate: "2020-03-10", + locationEncrypted: sealLocation(ESSEN), + }, + }); + // Entries: one before the source reaches, two nearby days on a 2020 trip, + // one before the home with no trip, one on a 2024 trip, one on a day with a + // weather row lacking air quality, one on a day already done, one at home + // without a row, a mood entry, and one inside the last week. + for (const day of [ + "2012-06-01", + "2020-03-01", + "2020-03-05", + "2020-06-01", + "2024-07-05", + "2025-05-01", + "2025-06-01", + "2025-09-10", + "2026-10-06", + ]) { + await measurementOn(day); + } + await prisma.moodEntry.create({ + data: { + userId: OWNER, + date: "2020-03-05", + mood: "GUT", + score: 4, + moodLoggedAt: new Date("2020-03-05T18:00:00Z"), + }, + }); + const sealedHome = sealLocation(BOCHUM); + await prisma.environmentContext.create({ + data: { + userId: OWNER, + date: "2025-05-01", + locationEncrypted: sealedHome, + source: "HOME", + tempMax: 20, + }, + }); + await prisma.environmentContext.create({ + data: { + userId: OWNER, + date: "2025-06-01", + locationEncrypted: sealedHome, + source: "HOME", + tempMax: 22, + pm25Mean: 7, + aqHours: 24, + aqFetchedAt: new Date("2025-06-02T00:00:00Z"), + }, + }); + fetchSpy = openMeteoStub(); + vi.stubGlobal("fetch", fetchSpy); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + vi.unstubAllEnvs(); +}); + +function requests() { + return fetchSpy.mock.calls.map(([input]) => new URL(String(input))); +} + +describe("air-quality history backfill", () => { + it("fills every past day with entries the source reaches, once", async () => { + const prisma = getPrismaClient(); + const result = await runAirQualityHistory(OWNER, { now: NOW }); + expect(result).toMatchObject({ + status: "complete", + total: 6, + done: 6, + remaining: 0, + }); + + const rows = await prisma.environmentContext.findMany({ + where: { userId: OWNER }, + orderBy: { date: "asc" }, + }); + expect(rows.map((r) => r.date)).toEqual([ + "2020-03-01", + "2020-03-05", + "2024-07-05", + "2025-05-01", + "2025-06-01", + "2025-09-10", + ]); + // Before the home and outside every period: no row, not counted. + expect(rows.find((r) => r.date === "2020-06-01")).toBeUndefined(); + for (const row of rows) expect(row.aqFetchedAt).not.toBeNull(); + const byDate = new Map(rows.map((r) => [r.date, r])); + // A day before the home was set, inside a period: at the period's place. + expect(byDate.get("2020-03-01")).toMatchObject({ + source: "TRAVEL", + pm25Mean: 12, + tempMax: 5, + lat: null, + }); + expect( + openLocation(byDate.get("2020-03-01")!.locationEncrypted!), + ).toMatchObject({ lat: 51.5, lon: 7.0 }); + // A day after the home was set, without a row: at the home. + const home = byDate.get("2025-09-10")!; + expect(home).toMatchObject({ source: "HOME", pm25Mean: 12, tempMax: 5 }); + expect(openLocation(home.locationEncrypted!)).toMatchObject({ + lat: 51.5, + lon: 7.2, + }); + // Pollen is not asked for before 2021, so it stays null, not zero. + expect(byDate.get("2020-03-01")!.pollenBirchMax).toBeNull(); + // A trip day: at the trip's place. + const trip = byDate.get("2024-07-05")!; + expect(trip.source).toBe("TRAVEL"); + expect(openLocation(trip.locationEncrypted!)).toMatchObject({ + lat: 38.7, + lon: -9.1, + }); + // A day with a weather row: air quality added, weather kept. + expect(byDate.get("2025-05-01")).toMatchObject({ + tempMax: 20, + pm25Mean: 12, + }); + // The done day was not touched. + expect(byDate.get("2025-06-01")!.pm25Mean).toBe(7); + + // The two 2020 days share one request each for air and weather; the + // coordinates are the coarse ones; the 2020 request asks the pollutant + // set only. The done day and the last week are never asked for. + const sent = requests(); + const air = sent.filter((u) => u.pathname === "/v1/air-quality"); + expect( + air.map((u) => [ + u.searchParams.get("start_date"), + u.searchParams.get("end_date"), + ]), + ).toEqual([ + ["2025-09-10", "2025-09-10"], + ["2025-05-01", "2025-05-01"], + ["2024-07-05", "2024-07-05"], + ["2020-03-01", "2020-03-05"], + ]); + const old = air.find( + (u) => u.searchParams.get("start_date") === "2020-03-01", + )!; + expect(old.searchParams.get("hourly")!.split(",")).toHaveLength(8); + for (const u of sent) { + expect(["51.5", "38.7"]).toContain(u.searchParams.get("latitude")); + } + // Weather only where a row was missing. + expect( + sent + .filter((u) => u.pathname === "/v1/archive") + .map((u) => u.searchParams.get("start_date")), + ).toEqual(["2025-09-10", "2024-07-05", "2020-03-01"]); + + const user = await prisma.user.findUniqueOrThrow({ where: { id: OWNER } }); + expect(user.environmentAqHistoryJson).toMatchObject({ + total: 6, + done: 6, + complete: true, + }); + + // Idempotent: a second run finds nothing to ask for. + fetchSpy.mockClear(); + const again = await runAirQualityHistory(OWNER, { now: NOW }); + expect(again).toMatchObject({ status: "complete", filled: 0, fetches: 0 }); + expect(fetchSpy).not.toHaveBeenCalled(); + }); + + it("stops at its own ceiling inside the account's daily share before sending", async () => { + const prisma = getPrismaClient(); + await prisma.rateLimit.create({ + data: { + key: `open-meteo-budget:account:${OWNER}:day`, + count: (AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS - 0.5) * 100, + resetAt: new Date(Date.now() + 3_600_000), + }, + }); + const result = await runAirQualityHistory(OWNER, { now: NOW }); + expect(result).toMatchObject({ status: "budget", filled: 0, done: 1 }); + expect(fetchSpy).not.toHaveBeenCalled(); + const user = await prisma.user.findUniqueOrThrow({ where: { id: OWNER } }); + expect(user.environmentAqHistoryJson).toMatchObject({ + total: 6, + done: 1, + complete: false, + }); + }); + + it("does nothing while the account or the operator has air quality off", async () => { + const prisma = getPrismaClient(); + await prisma.user.update({ + where: { id: OWNER }, + data: { environmentAirQualityEnabled: false }, + }); + expect((await runAirQualityHistory(OWNER, { now: NOW })).status).toBe( + "inactive", + ); + await prisma.user.update({ + where: { id: OWNER }, + data: { environmentAirQualityEnabled: true }, + }); + vi.stubEnv("ENVIRONMENT_AIR_QUALITY_DISABLED", "true"); + expect((await runAirQualityHistory(OWNER, { now: NOW })).status).toBe( + "inactive", + ); + expect(fetchSpy).not.toHaveBeenCalled(); + expect( + await prisma.environmentContext.count({ + where: { userId: OWNER, aqFetchedAt: { not: null } }, + }), + ).toBe(1); + }); + + it("runs through the queue handler, and discovery offers only a due account", async () => { + const sends: Array<{ name: string; data: unknown }> = []; + const boss = { + send: vi.fn(async (name: string, data: unknown) => { + sends.push({ name, data }); + return "job-id"; + }), + }; + const outcome = await handleEnvironmentAqHistory(boss as never, { + userId: OWNER, + }); + expect(outcome.ok).toBe(true); + const state = ( + await getPrismaClient().user.findUniqueOrThrow({ where: { id: OWNER } }) + ).environmentAqHistoryJson as { complete: boolean }; + expect(state.complete).toBe(true); + // Through and checked just now: discovery leaves it alone. + sends.length = 0; + await handleEnvironmentAqHistory(boss as never, {}); + expect(sends).toEqual([]); + }); +});