mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 13:01:00 +00:00
feat(insights): choose the status regeneration gap by ingest source
A sync from a watch or an integration now waits the instance's sync gap (AppSettings.aiSyncRegenGapHours, else AI_SYNC_REGEN_GAP_HOURS, else 6 h) before it may regenerate a status card again; a hand entry keeps the hour. Every caller of afterMeasurementMutation and invalidateStatusInsightsForTypes names its source, so a new ingest path cannot skip the choice. Refs #1217
This commit is contained in:
1 parent
f75a2fe343
commit
ae8b614dd0
47 files changed
+452
-165
No files matched your search
@@ -327,6 +327,13 @@ API_TOKEN_HMAC_KEY="<openssl rand -hex 32>"
|
||||
# docs/integrations/ai-providers.md.
|
||||
# DOCUMENT_AI_LIMIT_PER_HOUR=""
|
||||
|
||||
# Minimum hours between two regenerations of a status card after a sync
|
||||
# (optional). A watch or an integration delivers readings all day, and each
|
||||
# delivery would otherwise regenerate the cards it touches every hour. Hand
|
||||
# entries keep a one-hour gap. Default 6, whole hours from 1 to 48. The value
|
||||
# under Admin > AI wins over this one. See docs/integrations/ai-providers.md.
|
||||
# AI_SYNC_REGEN_GAP_HOURS=""
|
||||
|
||||
# Hourly upload ceiling for each document-upload token (optional). Applies
|
||||
# only to uploads made with a token minted under Settings > API & Tokens >
|
||||
# Document tokens: a Paperless workflow or the import script. Every upload
|
||||
|
||||
@@ -332,6 +332,11 @@ x-healthlog-env: &healthlog_env
|
||||
# (the `environment:` block is a whitelist). See
|
||||
# docs/integrations/ai-providers.md.
|
||||
DOCUMENT_AI_LIMIT_PER_HOUR: "${DOCUMENT_AI_LIMIT_PER_HOUR:-}"
|
||||
# Minimum hours between two status-card regenerations after a sync
|
||||
# (optional; default 6, whole hours 1-48; hand entries keep one hour). The
|
||||
# value under Admin > AI wins. Listed here so an operator value in .env
|
||||
# reaches the container. See docs/integrations/ai-providers.md.
|
||||
AI_SYNC_REGEN_GAP_HOURS: "${AI_SYNC_REGEN_GAP_HOURS:-}"
|
||||
# Hourly upload ceiling for each narrow document-upload token (a Paperless
|
||||
# workflow, the import script; optional; default 120, clamped 1-1000).
|
||||
# Uploads from the web app and the phone keep their own bucket. Listed here
|
||||
|
||||
@@ -431,6 +431,14 @@ many documents against a generous provider budget; lower it on a
|
||||
tight one. The 429 names the actual wait (the window is an hour) and
|
||||
carries the reset instant in `X-RateLimit-Reset`.
|
||||
|
||||
`AI_SYNC_REGEN_GAP_HOURS` (default `6`, whole hours from 1 to 48) is
|
||||
the minimum time between two regenerations of a status card after a sync.
|
||||
A watch or an integration delivers readings all day, and each delivery
|
||||
moves today's figures, so without a gap every touched card regenerated
|
||||
about once an hour. A reading entered by hand keeps a one-hour gap, so it
|
||||
is still narrated the same day. The value set under Admin > AI wins over
|
||||
the environment.
|
||||
|
||||
`LABS_OCR_LIMIT_PER_HOUR` (default `6`, clamped to 1–1000) does the
|
||||
same for lab-report scans. The photo or PDF scan and the browser-OCR
|
||||
text scan share one bucket per user, and a scan that fails before the
|
||||
|
||||
@@ -142,6 +142,10 @@
|
||||
"name": "DOCUMENT_AI_LIMIT_PER_HOUR",
|
||||
"purpose": "Per-user hourly ceiling for the document-AI actions (read/index, extract, suggest, summarise share one bucket). Default 6, clamped to 1-1000."
|
||||
},
|
||||
{
|
||||
"name": "AI_SYNC_REGEN_GAP_HOURS",
|
||||
"purpose": "Minimum hours between two status-card regenerations after a sync. Default 6, whole hours 1-48; hand entries keep one hour. The admin AI setting wins."
|
||||
},
|
||||
{
|
||||
"name": "DOCUMENT_UPLOAD_LIMIT_PER_HOUR",
|
||||
"purpose": "Hourly upload ceiling per document-upload token (Paperless workflow, import script). Default 120, clamped to 1-1000. Web and phone uploads keep their own bucket."
|
||||
|
||||
@@ -443,7 +443,10 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
// CSV pays at most ~N (type, day) recomputes, not 10 000 per-row hooks
|
||||
// — and the status re-warm for the touched types. Best-effort: a
|
||||
// populator hiccup never fails the importer.
|
||||
await afterMeasurementMutation(userId, touchedMeasurements, "import.csv");
|
||||
await afterMeasurementMutation(userId, touchedMeasurements, {
|
||||
source: "manual",
|
||||
label: "import.csv",
|
||||
});
|
||||
}
|
||||
|
||||
await auditLog("import.csv.upload", {
|
||||
|
||||
@@ -483,7 +483,12 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
// (type, day) the import touched (collapsed, so a 10 000-row restore pays
|
||||
// at most ~N recomputes) and re-warm the touched types' status
|
||||
// assessments. Best-effort: a populator hiccup never fails the importer.
|
||||
await afterMeasurementMutation(userId, touchedMeasurements, "import");
|
||||
await afterMeasurementMutation(userId, touchedMeasurements, {
|
||||
// A file the person chose to import: a deliberate act, narrated like a
|
||||
// hand entry rather than held to the sync gap.
|
||||
source: "manual",
|
||||
label: "import",
|
||||
});
|
||||
|
||||
annotate({
|
||||
meta: {
|
||||
|
||||
@@ -248,10 +248,14 @@ export const PUT = apiHandler(
|
||||
// NEW identity ride the list: an edit can move the row across day
|
||||
// boundaries or re-type it, and the helper collapses identical
|
||||
// (type, day) pairs so the common in-place edit fires once.
|
||||
await afterMeasurementMutation(user.id, [
|
||||
{ type: measurement.type, measuredAt: measurement.measuredAt },
|
||||
{ type: existing.type, measuredAt: existing.measuredAt },
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
user.id,
|
||||
[
|
||||
{ type: measurement.type, measuredAt: measurement.measuredAt },
|
||||
{ type: existing.type, measuredAt: existing.measuredAt },
|
||||
],
|
||||
{ source: "manual" },
|
||||
);
|
||||
|
||||
return apiSuccess(shapeMeasurementNotes(measurement));
|
||||
},
|
||||
@@ -312,9 +316,11 @@ export const DELETE = apiHandler(
|
||||
|
||||
// v1.37.19 (C2-F1) — shared post-mutation tail (the recompute drops
|
||||
// the rollup row when the day's measurement count goes to zero).
|
||||
await afterMeasurementMutation(user.id, [
|
||||
{ type: existing.type, measuredAt: existing.measuredAt },
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
user.id,
|
||||
[{ type: existing.type, measuredAt: existing.measuredAt }],
|
||||
{ source: "manual" },
|
||||
);
|
||||
|
||||
return apiSuccess({ deleted: true });
|
||||
},
|
||||
|
||||
@@ -1209,7 +1209,12 @@ async function postBatch(request: NextRequest): Promise<Response> {
|
||||
// v1.37.19 (C2-F1) — shared post-mutation tail for every distinct
|
||||
// (type, day) the batch touched. Updates participate as well as
|
||||
// inserts, so this deliberately retains the prepared-row scope.
|
||||
await afterMeasurementMutation(user.id, writtenIdentities);
|
||||
// A batch is the phone's sync, held to the sync gap, unless it wrote one
|
||||
// of the person's own hand-entered readings, which is narrated like any
|
||||
// other hand entry.
|
||||
await afterMeasurementMutation(user.id, writtenIdentities, {
|
||||
source: wroteManualRow ? "manual" : "sync",
|
||||
});
|
||||
|
||||
// S4 — Apple-Health sleep is the third sleep transport. If any SLEEP_DURATION
|
||||
// row for last night just landed, kick the debounced morning refresh so the
|
||||
|
||||
@@ -154,10 +154,12 @@ describe("POST /api/measurements/bulk-delete", () => {
|
||||
expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(3);
|
||||
// And one status re-warm over the distinct types.
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledTimes(1);
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledWith("user-1", [
|
||||
"WEIGHT",
|
||||
"PULSE",
|
||||
]);
|
||||
// A delete is the person's own act: narrated like a hand entry.
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledWith(
|
||||
"user-1",
|
||||
["WEIGHT", "PULSE"],
|
||||
{ source: "manual" },
|
||||
);
|
||||
});
|
||||
|
||||
it("is a no-op (deleted: 0) when no id is owned — no rollup, no invalidate", async () => {
|
||||
|
||||
@@ -123,7 +123,7 @@ async function postBulkDelete(request: NextRequest): Promise<Response> {
|
||||
type: row.type as MeasurementType,
|
||||
measuredAt: row.measuredAt,
|
||||
})),
|
||||
"measurements bulk-delete",
|
||||
{ source: "manual", label: "measurements bulk-delete" },
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -118,9 +118,12 @@ describe("DELETE /api/measurements/by-external-ids", () => {
|
||||
|
||||
// The deleted row's type is re-warmed; the zero-row case below is the
|
||||
// twin that must not reach it.
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledWith("user-1", [
|
||||
"WEIGHT",
|
||||
]);
|
||||
// The phone's HealthKit deletions arrive with its sync: the sync gap.
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledWith(
|
||||
"user-1",
|
||||
["WEIGHT"],
|
||||
{ source: "sync" },
|
||||
);
|
||||
});
|
||||
|
||||
it("is a no-op for a foreign-origin externalId that collides on a non-app-minted row", async () => {
|
||||
|
||||
@@ -169,7 +169,7 @@ async function deleteByExternalIds(request: NextRequest): Promise<Response> {
|
||||
|
||||
// v1.37.19 (C2-F1) — shared post-mutation tail (identities collected
|
||||
// BEFORE the destructive statement, while readable).
|
||||
await afterMeasurementMutation(user.id, affectedRows);
|
||||
await afterMeasurementMutation(user.id, affectedRows, { source: "sync" });
|
||||
}
|
||||
|
||||
return apiSuccess({ deletedCount: result.count });
|
||||
|
||||
@@ -149,10 +149,12 @@ describe("POST /api/measurements/restore", () => {
|
||||
expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(3);
|
||||
// And one status re-warm over the distinct types.
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledTimes(1);
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledWith("user-1", [
|
||||
"WEIGHT",
|
||||
"PULSE",
|
||||
]);
|
||||
// A restore is the person's own act: narrated like a hand entry.
|
||||
expect(invalidateStatusInsightsForTypes).toHaveBeenCalledWith(
|
||||
"user-1",
|
||||
["WEIGHT", "PULSE"],
|
||||
{ source: "manual" },
|
||||
);
|
||||
});
|
||||
|
||||
it("is a no-op (restored: 0) when no id is owned + tombstoned — no rollup, no invalidate", async () => {
|
||||
|
||||
@@ -122,7 +122,7 @@ async function postRestore(request: NextRequest): Promise<Response> {
|
||||
type: row.type as MeasurementType,
|
||||
measuredAt: row.measuredAt,
|
||||
})),
|
||||
"measurements restore",
|
||||
{ source: "manual", label: "measurements restore" },
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -963,6 +963,7 @@ async function postMeasurement(request: NextRequest) {
|
||||
await afterMeasurementMutation(
|
||||
user.id,
|
||||
results.map((r) => ({ type: r.type, measuredAt: r.measuredAt })),
|
||||
{ source: "manual" },
|
||||
);
|
||||
|
||||
return apiSuccess(shapedResults, 201);
|
||||
@@ -1116,9 +1117,11 @@ async function postMeasurement(request: NextRequest) {
|
||||
// pg-boss; WEEK / MONTH / YEAR recomputes are enqueued under the
|
||||
// hood. Best-effort — a populator hiccup never fails the user's
|
||||
// write.
|
||||
await afterMeasurementMutation(user.id, [
|
||||
{ type: measurement.type, measuredAt: measurement.measuredAt },
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
user.id,
|
||||
[{ type: measurement.type, measuredAt: measurement.measuredAt }],
|
||||
{ source: "manual" },
|
||||
);
|
||||
|
||||
return apiSuccess(shapeMeasurementNotes(measurement), 201);
|
||||
}
|
||||
@@ -1259,10 +1262,11 @@ async function postNamedSleepStage(
|
||||
).catch(() => {});
|
||||
void maybeEnqueueMorningRefresh(userId, [segment.end]).catch(() => {});
|
||||
}
|
||||
await afterMeasurementMutation(userId, [
|
||||
{ type, measuredAt: segment.end },
|
||||
...(verdict.dirtyIdentities ?? []),
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
userId,
|
||||
[{ type, measuredAt: segment.end }, ...(verdict.dirtyIdentities ?? [])],
|
||||
{ source: "manual" },
|
||||
);
|
||||
}
|
||||
|
||||
return namedSleepStageResponse(verdict.row.id, status, {
|
||||
|
||||
@@ -374,12 +374,16 @@ async function postAssessment(request: NextRequest): Promise<Response> {
|
||||
// route it through the shared post-mutation tail so the (type, day)
|
||||
// rollup bucket and the cached status assessment converge without
|
||||
// waiting for a nightly discovery pass.
|
||||
await afterMeasurementMutation(user.id, [
|
||||
{
|
||||
type: INSTRUMENT_MEASUREMENT_TYPE[id] as MeasurementType,
|
||||
measuredAt: when,
|
||||
},
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
user.id,
|
||||
[
|
||||
{
|
||||
type: INSTRUMENT_MEASUREMENT_TYPE[id] as MeasurementType,
|
||||
measuredAt: when,
|
||||
},
|
||||
],
|
||||
{ source: "manual" },
|
||||
);
|
||||
|
||||
// v1.27.6 — a screening can be planned as a Vorsorge reminder keyed on
|
||||
// PHQ9_SCORE / GAD7_SCORE. Kick the eventful satisfy worker so completing
|
||||
|
||||
@@ -900,6 +900,7 @@ export async function upsertFitbitMeasurements(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`fitbit: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -129,7 +129,9 @@ export async function syncUserFitbit(
|
||||
// the aggregator's `< to` upper bound covers it.
|
||||
const to = new Date(sorted[sorted.length - 1]! + 24 * 60 * 60 * 1000);
|
||||
await recomputeUserRollups(userId, { types, from, to });
|
||||
invalidateStatusInsightsForTypes(userId, types).catch((err) => {
|
||||
invalidateStatusInsightsForTypes(userId, types, {
|
||||
source: "sync",
|
||||
}).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`fitbit: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
);
|
||||
|
||||
@@ -1606,6 +1606,7 @@ export async function upsertGoogleHealthMeasurements(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`google-health: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -370,7 +370,9 @@ async function runGoogleHealthSyncCycle(
|
||||
const from = new Date(sorted[0]!);
|
||||
const to = new Date(sorted[sorted.length - 1]! + 24 * 60 * 60 * 1000);
|
||||
await recomputeUserRollups(userId, { types, from, to });
|
||||
invalidateStatusInsightsForTypes(userId, types).catch((err) => {
|
||||
invalidateStatusInsightsForTypes(userId, types, {
|
||||
source: "sync",
|
||||
}).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`google-health: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
);
|
||||
|
||||
@@ -19,6 +19,7 @@ const noteFindMany = vi.fn();
|
||||
const aiCapabilityForRecord = vi.fn();
|
||||
const enqueueStatusGeneration = vi.fn();
|
||||
const userFindUnique = vi.fn();
|
||||
const appSettingsFindUnique = vi.fn();
|
||||
const measurementFindMany = vi.fn();
|
||||
// v1.42 — type discovery goes through the loose-index-scan helper; the
|
||||
// rows it answers with are the types themselves.
|
||||
@@ -40,6 +41,9 @@ vi.mock("@/lib/db", () => ({
|
||||
findMany: (...a: unknown[]) => noteFindMany(...a),
|
||||
},
|
||||
user: { findUnique: (...a: unknown[]) => userFindUnique(...a) },
|
||||
appSettings: {
|
||||
findUnique: (...a: unknown[]) => appSettingsFindUnique(...a),
|
||||
},
|
||||
measurement: {
|
||||
findMany: (...a: unknown[]) => measurementFindMany(...a),
|
||||
},
|
||||
@@ -85,10 +89,19 @@ beforeEach(() => {
|
||||
});
|
||||
enqueueStatusGeneration.mockResolvedValue(undefined);
|
||||
userFindUnique.mockResolvedValue({ locale: "de" });
|
||||
appSettingsFindUnique.mockResolvedValue({ aiSyncRegenGapHours: null });
|
||||
measurementFindMany.mockResolvedValue([]);
|
||||
measurementGroupBy.mockResolvedValue([]);
|
||||
delete process.env.AI_SYNC_REGEN_GAP_HOURS;
|
||||
});
|
||||
|
||||
/** How far in the past the freshness probe's cutoff sat, in hours. */
|
||||
function probedGapHours(before: number): number {
|
||||
const where = noteFindMany.mock.calls[0][0].where;
|
||||
const cutoff = (where.generatedAt.gte as Date).getTime();
|
||||
return Math.round((before - cutoff) / (60 * 60 * 1000));
|
||||
}
|
||||
|
||||
/** Distinct scopes the invalidator enqueued a regenerate for. */
|
||||
function enqueuedScopes(): string[] {
|
||||
return [
|
||||
@@ -102,7 +115,9 @@ function enqueuedScopes(): string[] {
|
||||
|
||||
describe("invalidateStatusInsightsForTypes", () => {
|
||||
it("maps WEIGHT to weight + bmi + general", async () => {
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["bmi", "general", "weight"]);
|
||||
expect(
|
||||
enqueueStatusGeneration.mock.calls.every(
|
||||
@@ -112,7 +127,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
});
|
||||
|
||||
it("never deletes the cached assessment rows (v1.16.8 — hash gate decides)", async () => {
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
// The rows stay for stale-while-revalidate AND for the content-hash
|
||||
// gate the worker's forced regeneration runs; deleting them here would
|
||||
// force a full LLM regeneration even for a re-synced unchanged batch.
|
||||
@@ -120,18 +137,20 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
});
|
||||
|
||||
it("maps both blood-pressure components to blood-pressure + general", async () => {
|
||||
await invalidateStatusInsightsForTypes("u1", [
|
||||
"BLOOD_PRESSURE_SYS",
|
||||
"BLOOD_PRESSURE_DIA",
|
||||
]);
|
||||
await invalidateStatusInsightsForTypes(
|
||||
"u1",
|
||||
["BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"],
|
||||
{ source: "manual" },
|
||||
);
|
||||
expect(enqueuedScopes()).toEqual(["blood-pressure", "general"]);
|
||||
});
|
||||
|
||||
it("maps PULSE and RESTING_HEART_RATE to pulse + general (+ the resting-HR generic scope)", async () => {
|
||||
await invalidateStatusInsightsForTypes("u1", [
|
||||
"PULSE",
|
||||
"RESTING_HEART_RATE",
|
||||
]);
|
||||
await invalidateStatusInsightsForTypes(
|
||||
"u1",
|
||||
["PULSE", "RESTING_HEART_RATE"],
|
||||
{ source: "manual" },
|
||||
);
|
||||
// PULSE feeds the specialised pulse + general scopes; RESTING_HEART_RATE
|
||||
// feeds those too AND carries its own generic assessment card
|
||||
// (`metric:RESTING_HEART_RATE`), so all three are dirtied.
|
||||
@@ -146,16 +165,18 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
// BLOOD_GLUCOSE has no specialised assessment, but v1.8.7.1 gives it a
|
||||
// generic `metric:BLOOD_GLUCOSE` card, so a fresh reading dirties the
|
||||
// general overview AND that generic scope.
|
||||
await invalidateStatusInsightsForTypes("u1", ["BLOOD_GLUCOSE"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["BLOOD_GLUCOSE"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["general", "metric:BLOOD_GLUCOSE"]);
|
||||
});
|
||||
|
||||
it("dedupes scopes across a mixed batch", async () => {
|
||||
await invalidateStatusInsightsForTypes("u1", [
|
||||
"WEIGHT",
|
||||
"PULSE",
|
||||
"BLOOD_GLUCOSE",
|
||||
]);
|
||||
await invalidateStatusInsightsForTypes(
|
||||
"u1",
|
||||
["WEIGHT", "PULSE", "BLOOD_GLUCOSE"],
|
||||
{ source: "manual" },
|
||||
);
|
||||
// weight + bmi + general (WEIGHT) ∪ pulse + general (PULSE) ∪
|
||||
// general + metric:BLOOD_GLUCOSE (BLOOD_GLUCOSE)
|
||||
expect(enqueuedScopes()).toEqual([
|
||||
@@ -168,14 +189,16 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
});
|
||||
|
||||
it("is a no-op for an empty type set (no DB call)", async () => {
|
||||
await invalidateStatusInsightsForTypes("u1", []);
|
||||
await invalidateStatusInsightsForTypes("u1", [], { source: "manual" });
|
||||
expect(noteFindMany).not.toHaveBeenCalled();
|
||||
expect(enqueueStatusGeneration).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("enqueues nothing while statusText is unavailable for the record", async () => {
|
||||
aiCapabilityForRecord.mockResolvedValue(UNAVAILABLE);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT", "PULSE"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT", "PULSE"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(aiCapabilityForRecord).toHaveBeenCalledWith("u1", "statusText");
|
||||
expect(noteFindMany).not.toHaveBeenCalled();
|
||||
expect(enqueueStatusGeneration).not.toHaveBeenCalled();
|
||||
@@ -183,7 +206,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
|
||||
it("enqueues a debounced regenerate for every dirtied scope, the user's locale only", async () => {
|
||||
userFindUnique.mockResolvedValue({ locale: "de" });
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
// weight + bmi + general, each warmed once for the user's resolved locale.
|
||||
expect(enqueuedScopes()).toEqual(["bmi", "general", "weight"]);
|
||||
expect(enqueueStatusGeneration).toHaveBeenCalledTimes(3);
|
||||
@@ -197,7 +222,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
|
||||
it("resolves the user's locale (en) for the regenerate, never the unused one", async () => {
|
||||
userFindUnique.mockResolvedValue({ locale: "en" });
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
const locales = new Set(
|
||||
enqueueStatusGeneration.mock.calls.map(
|
||||
(c) => (c[0] as { locale: string }).locale,
|
||||
@@ -210,7 +237,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
// A steps sample touches only the general overview, never the weight /
|
||||
// pulse / bp scopes — the constant Apple-Health sync must not thrash
|
||||
// every card's cache.
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVITY_STEPS"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVITY_STEPS"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["general", "metric:STEPS"]);
|
||||
});
|
||||
|
||||
@@ -219,14 +248,18 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
// its own generic assessment card (`metric:BLOOD_GLUCOSE`), so the
|
||||
// freshly-synced metric's card refreshes in the background instead of
|
||||
// lagging until the nightly warm pass.
|
||||
await invalidateStatusInsightsForTypes("u1", ["BLOOD_GLUCOSE"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["BLOOD_GLUCOSE"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["general", "metric:BLOOD_GLUCOSE"]);
|
||||
});
|
||||
|
||||
it("maps a remapped MeasurementType to its registry metric id", async () => {
|
||||
// ACTIVE_ENERGY_BURNED is stored under that DB type but the generic
|
||||
// metric id (route param + cache scope) is ACTIVE_ENERGY.
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVE_ENERGY_BURNED"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVE_ENERGY_BURNED"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["general", "metric:ACTIVE_ENERGY"]);
|
||||
});
|
||||
|
||||
@@ -234,7 +267,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
// A weight sample feeds weight + bmi + general; WEIGHT is one of the
|
||||
// seven specialised metrics and has no generic registry entry, so no
|
||||
// `metric:` scope is enqueued.
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["bmi", "general", "weight"]);
|
||||
expect(enqueuedScopes().some((s) => s.startsWith("metric:"))).toBe(false);
|
||||
});
|
||||
@@ -249,7 +284,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
// `general` was regenerated within the gap; a fresh WEIGHT sample
|
||||
// must not re-enqueue it. weight + bmi are still stale, so they refresh.
|
||||
noteFindMany.mockResolvedValue([freshRow("general")]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["bmi", "weight"]);
|
||||
});
|
||||
|
||||
@@ -261,7 +298,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
// this is what lets same-day data be narrated same-day without
|
||||
// reopening the per-batch regeneration storm.
|
||||
noteFindMany.mockResolvedValue([]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueuedScopes()).toEqual(["bmi", "general", "weight"]);
|
||||
});
|
||||
|
||||
@@ -271,7 +310,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
freshRow("bmi"),
|
||||
freshRow("general"),
|
||||
]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(enqueueStatusGeneration).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
@@ -280,7 +321,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
// scope that recently stalled must retry rather than be debounced into
|
||||
// staying cold. The probe asks the database for noted rows only; with
|
||||
// none returned, every dirtied scope refreshes.
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
const where = noteFindMany.mock.calls[0][0].where;
|
||||
expect(where.textEncrypted).toEqual({ not: null });
|
||||
expect(enqueuedScopes()).toEqual(["bmi", "general", "weight"]);
|
||||
@@ -289,7 +332,9 @@ describe("invalidateStatusInsightsForTypes", () => {
|
||||
it("scopes the freshness probe to the user's resolved locale and a 1 h gap cutoff", async () => {
|
||||
userFindUnique.mockResolvedValue({ locale: "en" });
|
||||
const before = Date.now();
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
const where = noteFindMany.mock.calls[0][0].where;
|
||||
expect(where.userId).toBe("u1");
|
||||
// Only the en notes for the dirtied scopes are probed.
|
||||
@@ -387,3 +432,64 @@ describe("enqueueStatusRefillForUser", () => {
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
// #1217 — the gap is chosen by where the batch came from. A watch drips
|
||||
// syncs all day and every drip moves today's partial bucket, so the content
|
||||
// hash cannot hold them back; a hand entry still has to be narrated the
|
||||
// same day.
|
||||
describe("regeneration gap by ingest source", () => {
|
||||
it("holds a sync to six hours by default", async () => {
|
||||
const before = Date.now();
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVITY_STEPS"], {
|
||||
source: "sync",
|
||||
});
|
||||
expect(probedGapHours(before)).toBe(6);
|
||||
});
|
||||
|
||||
it("keeps a hand entry at one hour", async () => {
|
||||
const before = Date.now();
|
||||
await invalidateStatusInsightsForTypes("u1", ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
expect(probedGapHours(before)).toBe(1);
|
||||
expect(appSettingsFindUnique).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("takes the sync gap from AI_SYNC_REGEN_GAP_HOURS when the operator set no value", async () => {
|
||||
process.env.AI_SYNC_REGEN_GAP_HOURS = "12";
|
||||
const before = Date.now();
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVITY_STEPS"], {
|
||||
source: "sync",
|
||||
});
|
||||
expect(probedGapHours(before)).toBe(12);
|
||||
});
|
||||
|
||||
it("prefers the operator's stored gap over the environment", async () => {
|
||||
process.env.AI_SYNC_REGEN_GAP_HOURS = "12";
|
||||
appSettingsFindUnique.mockResolvedValue({ aiSyncRegenGapHours: 3 });
|
||||
const before = Date.now();
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVITY_STEPS"], {
|
||||
source: "sync",
|
||||
});
|
||||
expect(probedGapHours(before)).toBe(3);
|
||||
});
|
||||
|
||||
it("ignores an out-of-range environment value", async () => {
|
||||
process.env.AI_SYNC_REGEN_GAP_HOURS = "0";
|
||||
const before = Date.now();
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVITY_STEPS"], {
|
||||
source: "sync",
|
||||
});
|
||||
expect(probedGapHours(before)).toBe(6);
|
||||
});
|
||||
|
||||
it("falls back to the environment when the settings row cannot be read", async () => {
|
||||
process.env.AI_SYNC_REGEN_GAP_HOURS = "24";
|
||||
appSettingsFindUnique.mockRejectedValue(new Error("db down"));
|
||||
const before = Date.now();
|
||||
await invalidateStatusInsightsForTypes("u1", ["ACTIVITY_STEPS"], {
|
||||
source: "sync",
|
||||
});
|
||||
expect(probedGapHours(before)).toBe(24);
|
||||
});
|
||||
});
|
||||
@@ -131,12 +131,17 @@ describe("upsertScoreRow", () => {
|
||||
now: new Date("2026-06-02T03:00:00Z"),
|
||||
});
|
||||
|
||||
expect(afterMeasurementMutation).toHaveBeenCalledWith("u1", [
|
||||
{
|
||||
type: "STRAIN_SCORE",
|
||||
measuredAt: new Date("2026-06-01T12:00:00.000Z"),
|
||||
},
|
||||
]);
|
||||
expect(afterMeasurementMutation).toHaveBeenCalledWith(
|
||||
"u1",
|
||||
[
|
||||
{
|
||||
type: "STRAIN_SCORE",
|
||||
measuredAt: new Date("2026-06-01T12:00:00.000Z"),
|
||||
},
|
||||
],
|
||||
// A computed score is not a hand entry: it waits the sync gap.
|
||||
{ source: "sync", label: "score" },
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -105,9 +105,11 @@ export async function upsertScoreRow(
|
||||
// it through the shared post-mutation tail so the (type, day) rollup
|
||||
// bucket and the cached status assessment converge on the same tick that
|
||||
// wrote the score, instead of waiting for a discovery pass.
|
||||
await afterMeasurementMutation(args.userId, [
|
||||
{ type: args.type, measuredAt },
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
args.userId,
|
||||
[{ type: args.type, measuredAt }],
|
||||
{ source: "sync", label: "score" },
|
||||
);
|
||||
}
|
||||
|
||||
/** The tally every score pass returns. */
|
||||
|
||||
@@ -34,39 +34,88 @@ import { aiCapabilityForRecord } from "@/lib/ai/capabilities/gate";
|
||||
import type { MeasurementType } from "@/generated/prisma/client";
|
||||
|
||||
/**
|
||||
* Minimum gap between a scope's last cached (re)generation and the next
|
||||
* ingest-driven re-enqueue. A scope whose cached assessment was
|
||||
* (re)generated within this window is left intact on a fresh measurement
|
||||
* ingest — not re-enqueued.
|
||||
* Where a batch of fresh measurements came from, which decides how soon the
|
||||
* status cards it dirties may regenerate.
|
||||
*
|
||||
* A constantly-syncing client (Apple Health drips batches every few
|
||||
* minutes) used to delete + re-enqueue every dirtied scope on every batch,
|
||||
* and because each delete dropped the cache row the per-(user,scope,locale)
|
||||
* enqueue singleton (120 s) could not coalesce across batches that arrive
|
||||
* minutes apart — so STEPS / general / a metric card regenerated several
|
||||
* times an hour and the assessment felt "regenerated on every visit". This
|
||||
* window coalesces the storm: a genuinely stale scope (no fresh assessment)
|
||||
* still refreshes immediately, but a scope refreshed inside the window is
|
||||
* skipped, so a fresh assessment survives the day's sync drip.
|
||||
*
|
||||
* v1.16.8 — first widened from 30 min to 6 h to bound provider spend, but
|
||||
* a 6 h wall meant same-day data was not narrated same-day: the nightly
|
||||
* warm at 04:30 restarted the window, so a notable 08:00 reading never
|
||||
* re-enqueued its scopes and the card showed the pre-reading text until
|
||||
* the next nightly tick. The window is now a ONE-HOUR minimum gap, and
|
||||
* the budget role the 6 h wall carried moves to the worker's content-hash
|
||||
* gate: every ingest that lands past the gap re-enqueues the dirtied
|
||||
* scopes, the worker's forced run re-gathers the snapshot, and the gate
|
||||
* (`refreshUnchangedStatusInsight`) turns an unchanged snapshot into a
|
||||
* timestamp refresh with zero provider cost. So this clock only bounds
|
||||
* the SQL-gather frequency — at most one worker run per scope per hour
|
||||
* under a constant sync drip — while provider spend tracks actual data
|
||||
* change, which is what the gate exists to meter. Comparing the hash
|
||||
* inline at invalidation time was rejected: it would run the full
|
||||
* per-scope data gather inside the ingest path, which is exactly the
|
||||
* work the queue exists to keep off that path.
|
||||
* - `manual`: a reading the person entered or changed by hand (the web
|
||||
* form, the phone's entry sheet, a Telegram reply, an MCP write, a
|
||||
* delete or restore). It should be narrated the same day, so its gap
|
||||
* stays short.
|
||||
* - `sync`: rows a device or an integration delivered on its own schedule
|
||||
* (Apple Health, Health Connect, Withings, Oura, Fitbit, Polar, WHOOP,
|
||||
* Nightscout, Google Health, a file import, a computed score). A watch
|
||||
* drips these all day, and every drip moves today's partial bucket, so
|
||||
* the content-hash gate cannot catch them; only a longer gap does.
|
||||
*/
|
||||
const INGEST_INVALIDATE_MIN_GAP_MS = 60 * 60 * 1000;
|
||||
export type StatusIngestSource = "manual" | "sync";
|
||||
|
||||
/**
|
||||
* Minimum gap between a scope's last cached (re)generation and the next
|
||||
* ingest-driven re-enqueue, for a hand entry. A scope whose cached
|
||||
* assessment was (re)generated within the gap is left intact and not
|
||||
* re-enqueued.
|
||||
*
|
||||
* History: v1.9.0 introduced the window (30 min) because a constantly
|
||||
* syncing client re-enqueued every dirtied scope on every batch. v1.16.8
|
||||
* widened it to 6 h for every ingest, which was rolled back to 1 h because a
|
||||
* lone 08:00 reading after the 04:30 nightly warm was not narrated the same
|
||||
* day. The gap is now chosen by source: a hand entry keeps the hour, a sync
|
||||
* gets {@link DEFAULT_SYNC_REGEN_GAP_HOURS} (or the operator's value).
|
||||
*/
|
||||
const MANUAL_INGEST_MIN_GAP_MS = 60 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* The sync gap when neither the operator's setting
|
||||
* (`AppSettings.aiSyncRegenGapHours`) nor `AI_SYNC_REGEN_GAP_HOURS` names one.
|
||||
* Six hours turns a watch's ~16 regenerations per scope and day into ~3.
|
||||
*/
|
||||
export const DEFAULT_SYNC_REGEN_GAP_HOURS = 6;
|
||||
|
||||
/** The operator's sync gap bounds, shared with the admin setting. */
|
||||
export const SYNC_REGEN_GAP_HOURS_MIN = 1;
|
||||
export const SYNC_REGEN_GAP_HOURS_MAX = 48;
|
||||
|
||||
/** A gap in hours, or null when the value is not a usable whole hour count. */
|
||||
function parseGapHours(value: unknown): number | null {
|
||||
const n =
|
||||
typeof value === "number"
|
||||
? value
|
||||
: typeof value === "string" && value.trim() !== ""
|
||||
? Number(value)
|
||||
: NaN;
|
||||
if (!Number.isInteger(n)) return null;
|
||||
if (n < SYNC_REGEN_GAP_HOURS_MIN || n > SYNC_REGEN_GAP_HOURS_MAX) return null;
|
||||
return n;
|
||||
}
|
||||
|
||||
/**
|
||||
* The instance's sync gap in hours: the operator's admin setting, else
|
||||
* `AI_SYNC_REGEN_GAP_HOURS`, else six. An unreadable settings row falls back
|
||||
* to the environment, never to a shorter gap.
|
||||
*/
|
||||
export async function loadSyncRegenGapHours(): Promise<number> {
|
||||
let stored: number | null = null;
|
||||
try {
|
||||
const row = await prisma.appSettings.findUnique({
|
||||
where: { id: "singleton" },
|
||||
select: { aiSyncRegenGapHours: true },
|
||||
});
|
||||
stored = parseGapHours(row?.aiSyncRegenGapHours);
|
||||
} catch {
|
||||
stored = null;
|
||||
}
|
||||
return (
|
||||
stored ??
|
||||
parseGapHours(process.env.AI_SYNC_REGEN_GAP_HOURS) ??
|
||||
DEFAULT_SYNC_REGEN_GAP_HOURS
|
||||
);
|
||||
}
|
||||
|
||||
/** The regeneration gap for one ingest, in milliseconds. */
|
||||
async function ingestRegenGapMs(source: StatusIngestSource): Promise<number> {
|
||||
if (source === "manual") return MANUAL_INGEST_MIN_GAP_MS;
|
||||
return (await loadSyncRegenGapHours()) * 60 * 60 * 1000;
|
||||
}
|
||||
|
||||
/**
|
||||
* The seven per-metric assessment scopes. Each generator persists its
|
||||
@@ -130,6 +179,7 @@ function statusScopesForMeasurementType(
|
||||
export async function invalidateStatusInsightsForTypes(
|
||||
userId: string,
|
||||
types: Iterable<MeasurementType>,
|
||||
options: { source: StatusIngestSource },
|
||||
): Promise<void> {
|
||||
// One ordered set of every scope the batch dirties. The seven specialised
|
||||
// scopes are bare slugs (`weight`, `general`, …); the generic HealthKit
|
||||
@@ -184,14 +234,20 @@ export async function invalidateStatusInsightsForTypes(
|
||||
// only the redundant churn is removed. Past the gap the enqueue always
|
||||
// goes through — the worker's content-hash gate decides whether the
|
||||
// batch actually changed anything (see the constant's doc).
|
||||
const freshScopes = await findRecentlyWarmedScopes(userId, locale, scopes);
|
||||
const gapMs = await ingestRegenGapMs(options.source);
|
||||
const freshScopes = await findRecentlyWarmedScopes(
|
||||
userId,
|
||||
locale,
|
||||
scopes,
|
||||
gapMs,
|
||||
);
|
||||
const staleScopes = Array.from(scopes).filter(
|
||||
(scope) => !freshScopes.has(scope),
|
||||
);
|
||||
if (staleScopes.length === 0) {
|
||||
annotate({
|
||||
action: { name: "insights.status.invalidate.debounced" },
|
||||
meta: { skipped: scopes.size, refreshed: 0 },
|
||||
meta: { skipped: scopes.size, refreshed: 0, source: options.source },
|
||||
});
|
||||
return;
|
||||
}
|
||||
@@ -210,7 +266,11 @@ export async function invalidateStatusInsightsForTypes(
|
||||
|
||||
annotate({
|
||||
action: { name: "insights.status.invalidate.debounced" },
|
||||
meta: { skipped: freshScopes.size, refreshed: staleScopes.length },
|
||||
meta: {
|
||||
skipped: freshScopes.size,
|
||||
refreshed: staleScopes.length,
|
||||
source: options.source,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
@@ -268,7 +328,7 @@ export async function enqueueStatusRefillForUser(
|
||||
|
||||
/**
|
||||
* v1.9.0 — return the subset of `scopes` whose cached assessment for
|
||||
* `locale` was generated within `INGEST_INVALIDATE_MIN_GAP_MS` and is a
|
||||
* `locale` was generated within `gapMs` and is a
|
||||
* real (non-stub) assessment. Those scopes are skipped by the ingest
|
||||
* invalidator so a fresh assessment survives the sync drip.
|
||||
*
|
||||
@@ -281,8 +341,9 @@ async function findRecentlyWarmedScopes(
|
||||
userId: string,
|
||||
locale: SupportedLocale,
|
||||
scopes: ReadonlySet<InsightStatusScope>,
|
||||
gapMs: number,
|
||||
): Promise<Set<InsightStatusScope>> {
|
||||
const cutoff = new Date(Date.now() - INGEST_INVALIDATE_MIN_GAP_MS);
|
||||
const cutoff = new Date(Date.now() - gapMs);
|
||||
// Match exactly the cache actions for the candidate scopes in this locale.
|
||||
const candidateActions = Array.from(scopes, (scope) =>
|
||||
statusCacheAction(scope, locale),
|
||||
|
||||
@@ -605,7 +605,11 @@ export async function handleAppleHealthImport(
|
||||
if (importedTypes.length > 0) {
|
||||
// Registered so the test suite can await the detached handle.
|
||||
trackBackgroundTask(
|
||||
invalidateStatusInsightsForTypes(userId, importedTypes).catch((err) => {
|
||||
invalidateStatusInsightsForTypes(userId, importedTypes, {
|
||||
// An export file the person chose to import, narrated like any
|
||||
// other import rather than held to the sync gap.
|
||||
source: "manual",
|
||||
}).catch((err) => {
|
||||
console.warn(
|
||||
`[apple-health-import] status-insight invalidate failed for user ${userId}`,
|
||||
err,
|
||||
|
||||
@@ -228,7 +228,11 @@ export async function runHealthConnectImport(
|
||||
.map(([type]) => type as MeasurementType);
|
||||
if (touched.length > 0) {
|
||||
trackBackgroundTask(
|
||||
invalidateStatusInsightsForTypes(userId, touched).catch((err) => {
|
||||
invalidateStatusInsightsForTypes(userId, touched, {
|
||||
// An export file the person chose to import, like the Apple Health
|
||||
// import: narrated like a hand entry, not held to the sync gap.
|
||||
source: "manual",
|
||||
}).catch((err) => {
|
||||
console.warn(
|
||||
`[health-connect-import] status-insight invalidate failed for user ${userId}`,
|
||||
err,
|
||||
|
||||
@@ -478,7 +478,7 @@ describe("MCP writes — post-write tail", () => {
|
||||
expect(afterMeasurementMutation).toHaveBeenCalledWith(
|
||||
"u-1",
|
||||
[expect.objectContaining({ type: "WEIGHT" })],
|
||||
"mcp",
|
||||
{ source: "manual", label: "mcp" },
|
||||
);
|
||||
});
|
||||
|
||||
@@ -495,7 +495,7 @@ describe("MCP writes — post-write tail", () => {
|
||||
expect.objectContaining({ type: "BLOOD_PRESSURE_SYS" }),
|
||||
expect.objectContaining({ type: "BLOOD_PRESSURE_DIA" }),
|
||||
],
|
||||
"mcp",
|
||||
{ source: "manual", label: "mcp" },
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -355,7 +355,7 @@ export async function logMcpMeasurement(input: {
|
||||
await afterMeasurementMutation(
|
||||
input.userId,
|
||||
[{ type: input.type, measuredAt }],
|
||||
"mcp",
|
||||
{ source: "manual", label: "mcp" },
|
||||
);
|
||||
|
||||
await auditLog("mcp.write.measurement", {
|
||||
@@ -529,7 +529,7 @@ export async function logMcpBloodPressure(input: {
|
||||
{ type: "BLOOD_PRESSURE_SYS", measuredAt },
|
||||
{ type: "BLOOD_PRESSURE_DIA", measuredAt },
|
||||
],
|
||||
"mcp",
|
||||
{ source: "manual", label: "mcp" },
|
||||
);
|
||||
|
||||
await auditLog("mcp.write.blood_pressure", {
|
||||
|
||||
@@ -96,7 +96,7 @@ describe("logTelegramMeasurement", () => {
|
||||
expect(afterMeasurementMutation).toHaveBeenCalledWith(
|
||||
"user-1",
|
||||
[expect.objectContaining({ type: "WEIGHT" })],
|
||||
"telegram-measurement",
|
||||
{ source: "manual", label: "telegram-measurement" },
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -162,7 +162,7 @@ export async function logTelegramMeasurement(input: {
|
||||
await afterMeasurementMutation(
|
||||
input.userId,
|
||||
[{ type: input.type, measuredAt }],
|
||||
"telegram-measurement",
|
||||
{ source: "manual", label: "telegram-measurement" },
|
||||
);
|
||||
|
||||
return { status: "ok" };
|
||||
|
||||
@@ -364,7 +364,7 @@ export async function applyAppleHealthDistanceRepair(
|
||||
type: REPAIRED_TYPE,
|
||||
measuredAt: row.measuredAt,
|
||||
})),
|
||||
"repair-apple-health-distance",
|
||||
{ source: "sync", label: "repair-apple-health-distance" },
|
||||
);
|
||||
await client.auditLog.update({
|
||||
where: { id: auditId },
|
||||
|
||||
@@ -333,6 +333,7 @@ export async function upsertNightscoutEntries(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`nightscout: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -573,7 +573,10 @@ export async function retireOuraSleepRecords(
|
||||
where: { userId, id: { in: rows.map((r) => r.id) }, deletedAt: null },
|
||||
data: { deletedAt: new Date() },
|
||||
});
|
||||
await afterMeasurementMutation(userId, rows, "oura");
|
||||
await afterMeasurementMutation(userId, rows, {
|
||||
source: "sync",
|
||||
label: "oura",
|
||||
});
|
||||
invalidateUserMeasurements(userId);
|
||||
annotate({ meta: { oura_sleep_retired_rows: res.count } });
|
||||
return res.count;
|
||||
@@ -808,6 +811,7 @@ export async function upsertOuraMeasurements(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`oura: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -404,6 +404,7 @@ export async function upsertPolarMeasurements(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`polar: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -47,20 +47,24 @@ const day = (iso: string) => new Date(iso);
|
||||
|
||||
describe("afterMeasurementMutation", () => {
|
||||
it("recomputes one bucket per distinct (type, day) and invalidates each type once", async () => {
|
||||
await afterMeasurementMutation("user-1", [
|
||||
{
|
||||
type: "WEIGHT" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T08:00:00Z"),
|
||||
},
|
||||
{
|
||||
type: "WEIGHT" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T21:00:00Z"),
|
||||
},
|
||||
{
|
||||
type: "PULSE" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T09:00:00Z"),
|
||||
},
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
"user-1",
|
||||
[
|
||||
{
|
||||
type: "WEIGHT" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T08:00:00Z"),
|
||||
},
|
||||
{
|
||||
type: "WEIGHT" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T21:00:00Z"),
|
||||
},
|
||||
{
|
||||
type: "PULSE" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T09:00:00Z"),
|
||||
},
|
||||
],
|
||||
{ source: "sync" },
|
||||
);
|
||||
|
||||
// Two rows on the same (WEIGHT, day) collapse into one recompute.
|
||||
expect(mocks.recompute).toHaveBeenCalledTimes(2);
|
||||
@@ -76,13 +80,15 @@ describe("afterMeasurementMutation", () => {
|
||||
);
|
||||
|
||||
expect(mocks.invalidate).toHaveBeenCalledTimes(1);
|
||||
const [userId, types] = mocks.invalidate.mock.calls[0];
|
||||
const [userId, types, options] = mocks.invalidate.mock.calls[0];
|
||||
expect(userId).toBe("user-1");
|
||||
expect([...types].sort()).toEqual(["PULSE", "WEIGHT"]);
|
||||
// The caller's source reaches the invalidator, which picks the gap by it.
|
||||
expect(options).toEqual({ source: "sync" });
|
||||
});
|
||||
|
||||
it("is a no-op for an empty row set", async () => {
|
||||
await afterMeasurementMutation("user-1", []);
|
||||
await afterMeasurementMutation("user-1", [], { source: "manual" });
|
||||
expect(mocks.recompute).not.toHaveBeenCalled();
|
||||
expect(mocks.invalidate).not.toHaveBeenCalled();
|
||||
});
|
||||
@@ -99,7 +105,7 @@ describe("afterMeasurementMutation", () => {
|
||||
measuredAt: day("2026-05-10T08:00:00Z"),
|
||||
},
|
||||
],
|
||||
"test.surface",
|
||||
{ source: "manual", label: "test.surface" },
|
||||
);
|
||||
} finally {
|
||||
warn.mockRestore();
|
||||
@@ -113,12 +119,16 @@ describe("afterMeasurementMutation", () => {
|
||||
const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
try {
|
||||
await expect(
|
||||
afterMeasurementMutation("user-1", [
|
||||
{
|
||||
type: "WEIGHT" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T08:00:00Z"),
|
||||
},
|
||||
]),
|
||||
afterMeasurementMutation(
|
||||
"user-1",
|
||||
[
|
||||
{
|
||||
type: "WEIGHT" as MeasurementType,
|
||||
measuredAt: day("2026-05-10T08:00:00Z"),
|
||||
},
|
||||
],
|
||||
{ source: "manual" },
|
||||
),
|
||||
).resolves.toBeUndefined();
|
||||
// Let the fire-and-forget rejection settle inside the helper's catch.
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
@@ -22,16 +22,28 @@ import {
|
||||
collapseToTypeDayKeys,
|
||||
recomputeBucketsForMeasurement,
|
||||
} from "@/lib/rollups/measurement-rollups";
|
||||
import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation";
|
||||
import {
|
||||
invalidateStatusInsightsForTypes,
|
||||
type StatusIngestSource,
|
||||
} from "@/lib/insights/status-invalidation";
|
||||
import { trackBackgroundTask } from "@/lib/logging/background-tasks";
|
||||
import type { MeasurementType } from "@/generated/prisma/client";
|
||||
|
||||
export async function afterMeasurementMutation(
|
||||
userId: string,
|
||||
rows: Array<{ type: MeasurementType; measuredAt: Date }>,
|
||||
/** Log prefix identifying the calling surface in the warn lines. */
|
||||
label = "measurements",
|
||||
options: {
|
||||
/**
|
||||
* Where the rows came from: a hand entry (`manual`) or a device or
|
||||
* integration (`sync`). Decides how soon the status cards they dirty may
|
||||
* regenerate (`StatusIngestSource`).
|
||||
*/
|
||||
source: StatusIngestSource;
|
||||
/** Log prefix identifying the calling surface in the warn lines. */
|
||||
label?: string;
|
||||
},
|
||||
): Promise<void> {
|
||||
const label = options.label ?? "measurements";
|
||||
if (rows.length === 0) return;
|
||||
|
||||
// Collapse to the distinct `(type, day)` set BEFORE recomputing so a
|
||||
@@ -53,7 +65,9 @@ export async function afterMeasurementMutation(
|
||||
// write on a detached handle, and one that lands while a vitest worker is
|
||||
// closing its RPC channel fails the whole run. No-op outside test.
|
||||
trackBackgroundTask(
|
||||
invalidateStatusInsightsForTypes(userId, types).catch((err) => {
|
||||
invalidateStatusInsightsForTypes(userId, types, {
|
||||
source: options.source,
|
||||
}).catch((err) => {
|
||||
console.warn(`[${label}] status-insight invalidate failed`, err);
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -149,7 +149,7 @@ describe("syncUserBody — profile weight is not ingested (v1.16.11)", () => {
|
||||
expect(afterMeasurementMutation).toHaveBeenCalledWith(
|
||||
"user1",
|
||||
[legacyRow],
|
||||
"whoop-body",
|
||||
{ source: "sync", label: "whoop-body" },
|
||||
);
|
||||
expect(prismaMock.measurement.deleteMany).toHaveBeenCalledWith({
|
||||
where: {
|
||||
|
||||
@@ -144,7 +144,7 @@ describe("processWhoopNotification — fetch-by-id enqueue", () => {
|
||||
expect(afterMutation).toHaveBeenCalledWith(
|
||||
"user-1",
|
||||
[{ type: "SLEEP_DURATION", measuredAt }],
|
||||
"whoop.webhook",
|
||||
{ source: "sync", label: "whoop.webhook" },
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -78,7 +78,10 @@ export async function syncUserBody(
|
||||
});
|
||||
if (legacy.length > 0) {
|
||||
await prisma.measurement.deleteMany({ where: legacyWhere });
|
||||
await afterMeasurementMutation(userId, legacy, "whoop-body");
|
||||
await afterMeasurementMutation(userId, legacy, {
|
||||
source: "sync",
|
||||
label: "whoop-body",
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(
|
||||
|
||||
@@ -421,13 +421,13 @@ export async function upsertWhoopMeasurements(
|
||||
const invalidateTouchedTypes = (): void => {
|
||||
if (touchedTypes.size === 0) return;
|
||||
|
||||
void invalidateStatusInsightsForTypes(userId, [...touchedTypes]).catch(
|
||||
(err) => {
|
||||
getEvent()?.addWarning(
|
||||
`whoop: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
);
|
||||
},
|
||||
);
|
||||
void invalidateStatusInsightsForTypes(userId, [...touchedTypes], {
|
||||
source: "sync",
|
||||
}).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`whoop: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
);
|
||||
});
|
||||
};
|
||||
|
||||
for (
|
||||
|
||||
@@ -197,7 +197,7 @@ export async function processWhoopNotification(
|
||||
type: row.type as MeasurementType,
|
||||
measuredAt: row.measuredAt,
|
||||
})),
|
||||
"whoop.webhook",
|
||||
{ source: "sync", label: "whoop.webhook" },
|
||||
);
|
||||
if (resource === "workout") {
|
||||
// The Workout model carries no soft-delete column, so a deleted
|
||||
|
||||
@@ -550,6 +550,7 @@ export async function syncUserActivity(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`withings activity: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -528,6 +528,7 @@ export async function syncUserEcg(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`withings ecg: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -638,6 +638,7 @@ export async function syncUserSleep(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`withings sleep: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -457,6 +457,7 @@ export async function syncUserMeasurements(
|
||||
invalidateStatusInsightsForTypes(
|
||||
userId,
|
||||
keys.map((k) => k.type),
|
||||
{ source: "sync" },
|
||||
).catch((err) => {
|
||||
getEvent()?.addWarning(
|
||||
`withings: status-insight invalidate failed for ${userId}: ${err}`,
|
||||
|
||||
@@ -126,7 +126,9 @@ async function grantManage(input: {
|
||||
const enqueueFromSharedMutation: (request: NextRequest) => Promise<Response> =
|
||||
apiHandler(async () => {
|
||||
const { user } = await requireRecordAuth("manage", "medications");
|
||||
await invalidateStatusInsightsForTypes(user.id, ["WEIGHT"]);
|
||||
await invalidateStatusInsightsForTypes(user.id, ["WEIGHT"], {
|
||||
source: "manual",
|
||||
});
|
||||
return NextResponse.json({ data: { queued: true }, error: null });
|
||||
});
|
||||
|
||||
|
||||
@@ -116,9 +116,11 @@ describe("all-time summaries with DAY buckets older than the fold window", () =>
|
||||
await recomputeUserRollups(userId);
|
||||
// A backdated reading, as a manual entry or a sync of old samples writes
|
||||
// it: the hook folds that one day, window or not.
|
||||
await afterMeasurementMutation(userId, [
|
||||
{ type: "WEIGHT", measuredAt: old.at },
|
||||
]);
|
||||
await afterMeasurementMutation(
|
||||
userId,
|
||||
[{ type: "WEIGHT", measuredAt: old.at }],
|
||||
{ source: "manual" },
|
||||
);
|
||||
|
||||
await expectTrueFigures(userId, [64, 80]);
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user