Merge the document layout default and the air quality history into release/v1.42.0

This commit is contained in:
Marc-André Bombeck committed 2026-10-09 10:02:56 +02:00
commit 57c44006ee
35 files changed
+1884 -76

No files matched your search

+18 -3
View File
@@ -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
+32 -1
View File
@@ -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:
+2
View File
@@ -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})",
+2
View File
@@ -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})",
+2
View File
@@ -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})",
+2
View File
@@ -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})",
+2
View File
@@ -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})",
+2
View File
@@ -5829,6 +5829,8 @@
"operatorDisabled": "이 서버에서는 대기질 기능이 꺼져 있습니다.",
"coverage": "대기질 데이터가 있는 날 {count}일",
"pollenEuropeOnly": "꽃가루 데이터는 유럽만 제공됩니다.",
"historyProgress": "대기질 과거 데이터 채우는 중: {total}일 중 {done}일",
"historyComplete": "출처가 제공하는 기록이 있는 이전 날짜 모두에 대기질이 채워졌습니다({total}일). 그보다 오래된 날짜는 비어 있습니다.",
"attribution": "데이터 출처와 라이선스",
"saveError": "설정을 저장하지 못했습니다.",
"latest": "가장 최근 날짜 ({date})",
+2
View File
@@ -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})",
@@ -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;
+6
View File
@@ -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")
@@ -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);
+14 -10
View File
@@ -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<DocumentsLayout> {
async function readStoredLayout(userId: string): Promise<unknown> {
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({
+4
View File
@@ -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" } });
@@ -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({
+18 -1
View File
@@ -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
@@ -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;
@@ -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")}`
: ""}
</p>
{data.airQuality.history && data.airQuality.history.total > 0 && (
<p
className="text-muted-foreground text-xs"
data-testid="environment-aq-history"
data-complete={data.airQuality.history.complete}
>
{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,
},
)}
</p>
)}
{data.latestDay?.airQuality && (
<div className="space-y-2">
<p className="text-sm font-medium">
+1
View File
@@ -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",
+24 -4
View File
@@ -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<Pick<DocumentsLayout, "view" | "arrangement">> {
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 }
: {}),
};
}
@@ -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);
});
});
@@ -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/");
+589
View File
@@ -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<string, unknown>;
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<string[]> {
// 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<Array<{ day: string }>>`
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<HistoryWork | null> {
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<AirQualityHistoryRunResult> {
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<string, HistoryDay> }
>();
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<string, DailyAirQualityObservation>(
air.days.map((d) => [d.date, d]),
);
const airFor = (date: string) => airByDay.get(date) ?? uncoverableDay(date);
let weather: Map<string, DailyEnvironmentObservation> | 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<unknown>[] = [];
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 };
}
+66 -33
View File
@@ -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<AirQualityFetchResult> {
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" };
}
+12 -4
View File
@@ -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<DailyEnvironmentObservation[]> {
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();
}
+36
View File
@@ -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<Record<BudgetWindow["name"], number>>;
/**
* 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<BudgetUsage> {
const keys = new Map<string, BudgetWindow["name"]>(
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;
}
+14 -10
View File
@@ -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<ArrayBuffer>,
@@ -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.
*/
+3
View File
@@ -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
@@ -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<boolean> {
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<JobOutcome> {
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,
});
}
+3
View File
@@ -466,4 +466,7 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
"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.",
),
};
@@ -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<EnvironmentAqHistoryPayload>(
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<void> {
);
}
// 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 } =
+1
View File
@@ -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" },
+1 -1
View File
@@ -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({
+23
View File
@@ -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
@@ -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<string, unknown> = { time };
for (const v of variables) hourly[v] = time.map(() => 12);
return Response.json({ hourly });
}
const daily: Record<string, unknown> = { 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<string, unknown> = { 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<typeof openMeteoStub>;
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([]);
});
});