diff --git a/src/lib/environment/service.ts b/src/lib/environment/service.ts index 2208e547d..d1f168127 100644 --- a/src/lib/environment/service.ts +++ b/src/lib/environment/service.ts @@ -230,25 +230,31 @@ export async function fetchAndStoreEnvironment(args: { } fetches += 1; - const obsByDate = new Map(observations.map((o) => [o.date, o])); const wanted = new Set(groupDays); - for (const obs of observations) { - if (!wanted.has(obs.date)) continue; - await upsertObservation(userId, loc, obs); - stored += 1; + // Batch the group's day upserts into a single transaction round-trip + // rather than awaiting one upsert per day in series (a 730-day backfill + // would otherwise issue 730 sequential statements). One transaction per + // location group keeps the per-backfill DB chatter bounded while + // preserving the idempotent upsert semantics. + const ops = observations + .filter((obs) => wanted.has(obs.date)) + .map((obs) => buildUpsert(userId, loc, obs)); + if (ops.length > 0) { + await prisma.$transaction(ops); + stored += ops.length; } // Days the feed did not return (e.g. beyond the settling lag) stay absent. - void obsByDate; } return { stored, skipped, fetches }; } -async function upsertObservation( +/** Build the idempotent per-day upsert operation (un-awaited, for a batch). */ +function buildUpsert( userId: string, loc: ResolvedLocation, obs: DailyEnvironmentObservation, -): Promise { +) { const data = { lat: loc.lat, lon: loc.lon, @@ -268,7 +274,7 @@ async function upsertObservation( weatherCode: obs.weatherCode, fetchedAt: new Date(), }; - await prisma.environmentContext.upsert({ + return prisma.environmentContext.upsert({ where: { userId_date: { userId, date: obs.date } }, create: { userId, date: obs.date, ...data }, update: data, diff --git a/src/lib/jobs/environment-fetch.ts b/src/lib/jobs/environment-fetch.ts index 97e865ecb..b21d8379d 100644 --- a/src/lib/jobs/environment-fetch.ts +++ b/src/lib/jobs/environment-fetch.ts @@ -20,7 +20,12 @@ import { prisma } from "@/lib/db"; import type { PgBoss } from "pg-boss"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; -import { isModuleEnabled } from "@/lib/modules/gate"; +import { + getOperatorModuleAvailability, + isModuleEnabled, + normalisePrefs, + resolveModuleEnabled, +} from "@/lib/modules/gate"; import { ENVIRONMENT_LOOKBACK_DAYS, ENVIRONMENT_MAX_BACKFILL_DAYS, @@ -96,17 +101,35 @@ export async function enqueueEnvironmentFetchDiscovery( boss: PgBoss, ): Promise<{ enqueued: number; skipped: number }> { // Candidate set: a home location is required for any non-override day, so an - // account without one cannot produce a row. The per-candidate module check - // then enforces the opt-in flag + operator availability. - const candidates = await prisma.user.findMany({ - where: { homeLat: { not: null }, homeLon: { not: null } }, - select: { id: true }, - }); + // account without one cannot produce a row. The opt-in flag + operator + // availability are then resolved in-memory from the candidates' preference + // blobs — one operator-availability read + one `findMany`, instead of an + // `isModuleEnabled` round-trip per candidate (N serial reads on a large + // instance). `environment` is opt-in and non-delegated, so the resolver only + // consults the per-user preference map + the operator availability. + const [candidates, operatorAvailability] = await Promise.all([ + prisma.user.findMany({ + where: { homeLat: { not: null }, homeLon: { not: null } }, + select: { id: true, modulePreferencesJson: true }, + }), + getOperatorModuleAvailability(), + ]); let enqueued = 0; let skipped = 0; - for (const { id } of candidates) { - if (!(await isModuleEnabled(id, "environment"))) { + for (const { id, modulePreferencesJson } of candidates) { + const isEnabled = resolveModuleEnabled( + "environment", + { + gender: null, + disableCoach: false, + modulePreferences: normalisePrefs(modulePreferencesJson), + cycleTrackingEnabled: null, + }, + false, + operatorAvailability, + ); + if (!isEnabled) { skipped += 1; continue; }