Reach every cached reader from the integration and context write paths

Integration syncs (Fitbit, Oura, Polar, WHOOP, Strava, Nightscout) now
mark the per-user analytics cells stale and sweep the workouts cache at
their sync tails — imported rows used to stay invisible to the cached
dashboard, analytics and workout readers until the TTL lapsed. The WHOOP
per-resource webhook/cron driver and the Polar workouts leg carry the
same call, matching the posture Google Health and the workout batch
route already had.

Illness, cycle and encounter write routes hard-evict the analytics
bucket through the new invalidateUserHealthContext helper: Rest Mode,
the cycle context and the Today-rail appointment feed all read from
cached cells under that bucket, and an interactive write must be true on
the very next read.
This commit is contained in:
Marc-André Bombeck committed 2026-08-13 18:33:46 +02:00
1 parent e3169fe3b3
commit 002d120b69
22 files changed
+162

No files matched your search

+5
View File
@@ -18,6 +18,7 @@
import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { apiHandler, requireAuth } from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
@@ -105,6 +106,10 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
details: cleared,
});
// The purge removes every row the cached cycle context was built on —
// evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: { name: "cycle.purge" },
meta: cleared,
+5
View File
@@ -12,6 +12,7 @@
import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
@@ -78,6 +79,10 @@ export const DELETE = apiHandler(
},
});
// A removed cycle changes the cycle context the cached analytics /
// snapshot / digest cells carry — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "cycle.cycle.delete",
+9
View File
@@ -11,6 +11,7 @@
import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
@@ -204,6 +205,10 @@ export const PATCH = apiHandler(
},
});
// A day-log edit changes the cycle context the cached analytics /
// snapshot / digest cells carry — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "cycle.day-log.update",
@@ -282,6 +287,10 @@ export const DELETE = apiHandler(
},
});
// A removed day (possibly with its cycle) changes the cycle context the
// cached analytics / snapshot / digest cells carry — evict the bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "cycle.day-log.delete",
+7
View File
@@ -25,6 +25,7 @@ import {
sanitiseZodIssues,
} from "@/lib/api-response";
import { prisma } from "@/lib/db";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { withIdempotency } from "@/lib/idempotency";
import { checkRateLimit } from "@/lib/rate-limit";
import { requireCycleEnabled } from "@/lib/cycle/gate";
@@ -228,6 +229,12 @@ async function postBulk(request: NextRequest): Promise<Response> {
},
});
// Landed rows change the cycle context the cached analytics / snapshot /
// digest cells carry — evict the analytics bucket.
if (inserted > 0 || updated > 0) {
invalidateUserHealthContext(user.id);
}
annotate({
action: { name: "cycle.bulk.ingest" },
meta: {
+5
View File
@@ -14,6 +14,7 @@
import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
@@ -173,6 +174,10 @@ async function postDayLog(request: NextRequest): Promise<Response> {
},
});
// A day-log write changes the cycle context the cached analytics /
// snapshot / digest cells carry — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "cycle.day-log.upsert",
+5
View File
@@ -12,6 +12,7 @@
import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
@@ -196,6 +197,10 @@ async function postPeriod(request: NextRequest): Promise<Response> {
},
});
// A period boundary changes the cycle context the cached analytics /
// snapshot / digest cells carry — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "cycle.period.boundary",
@@ -14,6 +14,7 @@ import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import { apiSuccess, apiError, getClientIp } from "@/lib/api-response";
@@ -92,6 +93,10 @@ export const POST = apiHandler(
details: { encounterId: id, occurredAt: row.occurredAt.toISOString() },
});
// A restored visit must reach the cached daily-digest / snapshot cells
// (Today rail) on the next read — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "encounter.visit.restore",
+9
View File
@@ -18,6 +18,7 @@ import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import { overwriteDetails } from "@/lib/sharing/audit-details";
@@ -342,6 +343,10 @@ export const PATCH = apiHandler(
},
});
// A rescheduled / re-statused visit must reach the cached daily-digest /
// snapshot cells (Today rail) on the next read — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "encounter.visit.update",
@@ -392,6 +397,10 @@ export const DELETE = apiHandler(
},
});
// A removed visit must leave the cached daily-digest / snapshot cells
// (Today rail) on the next read — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "encounter.visit.delete",
+5
View File
@@ -15,6 +15,7 @@ import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import {
@@ -299,6 +300,10 @@ async function postEncounter(request: NextRequest): Promise<Response> {
details: { encounterId: created.id, status: created.status },
});
// A booked visit must reach the cached daily-digest / snapshot cells
// (Today rail) on the next read — evict the analytics bucket.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "encounter.visit.create",
@@ -11,6 +11,7 @@ import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import {
@@ -94,6 +95,10 @@ export const PATCH = apiHandler(
details: { episodeId: id },
});
// Resolving flips Rest Mode off — evict the cached analytics / snapshot /
// digest cells so the context is true on the next read.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "illness.episode.resolve",
@@ -14,6 +14,7 @@ import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import { apiSuccess, apiError, getClientIp } from "@/lib/api-response";
@@ -55,6 +56,9 @@ export const POST = apiHandler(
ipAddress: getClientIp(request),
details: { episodeId: id },
});
// Restoring an unresolved episode can flip Rest Mode back on — evict
// the cached analytics / snapshot / digest cells.
invalidateUserHealthContext(user.id);
}
const restored = await prisma.illnessEpisode.findUniqueOrThrow({
@@ -11,6 +11,7 @@ import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import { overwriteDetails } from "@/lib/sharing/audit-details";
@@ -184,6 +185,11 @@ export const PATCH = apiHandler(
},
});
// An onset / lifecycle / resolvedAt edit can flip Rest Mode — evict the
// cached analytics / snapshot / digest cells so the context is true on
// the next read.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "illness.episode.update",
@@ -224,6 +230,9 @@ export const DELETE = apiHandler(
ipAddress: getClientIp(request),
details: { episodeId: id },
});
// Deleting an active episode can flip Rest Mode off — evict the cached
// analytics / snapshot / digest cells.
invalidateUserHealthContext(user.id);
}
annotate({
+5
View File
@@ -14,6 +14,7 @@ import { NextRequest } from "next/server";
import { prisma } from "@/lib/db";
import { apiHandler, requireRecordAuth } from "@/lib/api-handler";
import { invalidateUserHealthContext } from "@/lib/cache/invalidate";
import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import {
@@ -135,6 +136,10 @@ async function postEpisode(request: NextRequest): Promise<Response> {
details: { episodeId: created.id, type: created.type },
});
// A new episode can flip Rest Mode on — evict the cached analytics /
// snapshot / digest cells so the context is true on the next read.
invalidateUserHealthContext(user.id);
annotate({
action: {
name: "illness.episode.create",
+13
View File
@@ -271,6 +271,19 @@ export function invalidateUserMedicationListLayout(userId: string): void {
caches.medicationListLayout.deleteByPrefix(userId);
}
/**
* Invalidate the analytics bucket after an illness / cycle / encounter write.
* `resolveRestMode` (active-episode context) feeds both the analytics report
* and the dashboard snapshot, and the daily-digest extras cell lives under the
* same bucket — all keyed `${userId}|…`. Hard-evict, mirroring the mood-write
* posture: these are interactive single-entry writes, and a marked-stale SWR
* cell would hand the user back the pre-write rest-mode / appointment context
* on the very next read.
*/
export function invalidateUserHealthContext(userId: string): void {
caches.analytics.deleteByPrefix(`${userId}|`);
}
/** Invalidate surfaces whose server decision includes pattern dismissal. */
export function invalidateUserCorrelationPatterns(userId: string): void {
caches.analytics.deleteByPrefix(`${userId}|`);
+9
View File
@@ -1,5 +1,6 @@
/** Fitbit sync orchestration and bounded cohort polling. */
import pLimit from "p-limit";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { prisma } from "@/lib/db";
import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status";
@@ -161,6 +162,14 @@ export async function syncUserFitbit(
await recordSyncSuccess(userId, "fitbit");
}
// Background-sync posture: mark the analytics cells stale (never hard-evict
// from a poll) and sweep the workouts cache so the imported rows reach the
// cached readers. Fires on a partial failure too — rows that DID land must
// not stay invisible for the rest of the TTL window.
if (total > 0) {
invalidateUserMeasurements(userId);
}
annotate({
action: { name: "fitbit.sync", details: { imported: total, failed } },
});
+7
View File
@@ -5,6 +5,7 @@
* schedules, and boss.work registrations.
*/
import { type Job } from "pg-boss";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { fireAndForget } from "@/lib/logging/fire-and-forget";
import { recordError } from "@/lib/jobs/worker-status";
import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
@@ -124,6 +125,12 @@ export async function runWhoopResourceSync(
// v1.18.1 — a fresh reading landed; resolve this user's Vorsorge
// reminders eventfully. Fire-and-forget; the cron is the net.
if (imported > 0) {
// Background-sync posture: mark the analytics cells stale (never
// hard-evict from a poll) and sweep the workouts cache so the
// imported rows reach the cached readers. This driver is the only
// sync tail the webhook / per-resource cron paths pass through —
// `syncUserWhoop` covers the manual + backfill paths itself.
invalidateUserMeasurements(userId);
fireAndForget(enqueueReminderSatisfy(userId), {
action: "reminder.satisfy.enqueue",
});
+9
View File
@@ -40,6 +40,7 @@ import {
emitInsertedMeasurementArrivals,
type InsertedMeasurementArrivalRow,
} from "@/lib/arrivals/measurement-emit";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
import {
fetchSgvEntries,
@@ -185,6 +186,14 @@ export async function syncUserNightscout(
entries,
);
// Background-sync posture: mark the analytics cells stale (never hard-evict
// from a poll) so the inserted rows reach the cached readers. Fires on a
// partial failure too — rows that DID land must not stay invisible for the
// rest of the TTL window.
if (inserted > 0) {
invalidateUserMeasurements(userId);
}
if (failedRows > 0) {
// Partial write failure: keep the tick honest. `transient` because the
// ~48 h re-fetch window self-heals a transient DB hiccup, while a
+9
View File
@@ -51,6 +51,7 @@ import {
emitInsertedMeasurementArrivals,
type InsertedMeasurementArrivalRow,
} from "@/lib/arrivals/measurement-emit";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
import type { SyncWriteResult } from "@/lib/outcome/written-outcome";
import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
@@ -498,6 +499,14 @@ export async function syncUserOura(
() => {},
);
// Background-sync posture: mark the analytics cells stale (never hard-evict
// from a poll) so the imported rows reach the cached readers. Fires on a
// partial failure too — rows that DID land must not stay invisible for the
// rest of the TTL window.
if (imported > 0) {
invalidateUserMeasurements(userId);
}
if (result.failures.length > 0) {
// Partial failure: keep the cycle honest. Record the failure and do NOT
// stamp success, so the freshness surface reflects that some collections
+8
View File
@@ -30,6 +30,7 @@
*/
import { prisma } from "@/lib/db";
import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { annotate, getEvent } from "@/lib/logging/context";
import {
markSyncFailureRecorded,
@@ -102,6 +103,13 @@ export async function syncUserPolarWorkouts(userId: string): Promise<number> {
throw markSyncFailureRecorded(err);
}
// Sweep the workouts cache (and mark the analytics cells stale — workouts
// feed achievements + analytics) so the upserted rows reach the cached
// readers, matching the `/api/workouts/batch` and Google Health writers.
if (imported > 0) {
invalidateUserMeasurements(userId);
}
return imported;
}
+9
View File
@@ -50,6 +50,7 @@ import {
collapseToTypeDayKeys,
recomputeBucketsForMeasurement,
} from "@/lib/rollups/measurement-rollups";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
import type { SyncWriteResult } from "@/lib/outcome/written-outcome";
import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
@@ -276,6 +277,14 @@ export async function syncUserPolar(userId: string): Promise<SyncWriteResult> {
() => {},
);
// Background-sync posture: mark the analytics cells stale (never hard-evict
// from a poll) so the imported rows reach the cached readers. Fires on a
// partial failure too — rows that DID land must not stay invisible for the
// rest of the TTL window.
if (imported > 0) {
invalidateUserMeasurements(userId);
}
if (failures.length > 0) {
// Partial failure: import what the healthy collections returned, but keep
// the cycle honest — record ONE failure and do NOT stamp success, so the
+9
View File
@@ -25,6 +25,7 @@
*/
import { prisma } from "@/lib/db";
import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { annotate, getEvent } from "@/lib/logging/context";
import {
markSyncFailureRecorded,
@@ -244,6 +245,14 @@ export async function syncUserStrava(
}
await recordSyncSuccess(userId, "strava");
// Background-sync posture: sweep the workouts cache and mark the analytics
// cells stale (workouts feed achievements + analytics) so the upserted rows
// reach the cached readers, matching the `/api/workouts/batch` writer.
if (imported > 0) {
invalidateUserMeasurements(userId);
}
annotate({
action: { name: "strava.sync.complete", details: { imported } },
});
+11
View File
@@ -5,6 +5,7 @@
* `sync-core.ts`. Resource leaves depend only on that core; this parent owns
* the static leaf graph and the full per-user sequencing policy.
*/
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { getEvent } from "@/lib/logging/context";
import {
isReauthRequired,
@@ -99,5 +100,15 @@ export async function syncUserWhoop(
if (!anyFailed && !allSoftSkipped) {
await recordSyncSuccess(userId, "whoop");
}
// Background-sync posture: mark the analytics cells stale (never hard-evict
// from a poll) and sweep the workouts cache so the imported rows reach the
// cached readers. Fires on a partial failure too — rows that DID land must
// not stay invisible for the rest of the TTL window. The per-resource
// webhook/cron jobs run their own call in `runWhoopResourceSync`.
if (total > 0) {
invalidateUserMeasurements(userId);
}
return { imported: total, failed: anyFailed || allSoftSkipped };
}