refactor(medications): drop the compliance rollup table that nothing reads

medication_compliance_rollups lost its last reader when compliance moved
to the schedule-anchored computation, yet every dose write still upserted
it, a boot backfill refolded it, and the wipe and backup plans carried it.

The per-dose hook keeps the one derived row that still hangs off a dose
day, the supplement nutrients, under dose-day-refresh. The pending-row
mints (today projection, reminder) no longer call it, since an untaken
dose adds no nutrients. The import job finishes without its per-day
recompute phase; its progress keeps touchedDays and rollupProcessed on
the wire, inert and documented as deprecated. The backfill queue is
retired on boot and migration 0414 drops the table under a bounded lock
wait.
This commit is contained in:
Marc-André Bombeck committed 2026-10-20 13:01:45 +02:00
1 parent b90c82cba3
commit 41de0ca126
71 files changed
+314 -2027

No files matched your search

+8 -6
View File
@@ -9544,10 +9544,9 @@ paths:
defend against a poisoned column. Audit-logged with the old and new zone.
A change is not only a display preference. The medication-compliance rollup keys its rows by the LOCAL day a
dose fell on, minted under whatever zone was in force at write time, so moving the zone re-buckets that history
and the stored keys go stale — and the coverage probe compares counts rather than keys, so it cannot notice. The
route therefore enqueues a re-fold of the trailing window whenever the zone actually changes. Best-effort and
A change is not only a display preference. The supplement nutrient rows key on the LOCAL day a dose fell on,
minted under whatever zone was in force at write time, so moving the zone re-buckets them. The route therefore
enqueues a recompute of the account's supplement nutrients whenever the zone actually changes. Best-effort and
de-duplicated, so rapid re-picks collapse into one.
requestBody:
required: true
@@ -52993,12 +52992,15 @@ components:
type: array
items:
type: string
description: Local calendar days (YYYY-MM-DD) touched so far, driving the chunked rollup recompute behind the write.
description: "Deprecated and inert: always `[]` on a job admitted from v1.43.8, when the per-day compliance rollup it
drove was removed. A job admitted earlier keeps the days it recorded. Kept for clients that decode it; do
not read it."
rollupProcessed:
type: integer
minimum: 0
maximum: 9007199254740991
description: Of `touchedDays`, how many have had their rollup recomputed.
description: "Deprecated and inert: always `0` on a job admitted from v1.43.8. Kept for clients that decode it; do not
read it."
required:
- processed
- total
+6 -10
View File
@@ -33,9 +33,9 @@ metadata (`medication_schedules`):
- After upgrading to v1.15.19+, when a user reports a compliance rate above
100 %, duplicate entries in the medication history, or a dose shown as
taken days away from its slot.
- The compliance rollup counts DISTINCT slots since v1.15.19, so the
percentages self-correct on recompute even without running this script.
Running it additionally clears the phantom rows from the history view.
- Compliance counts DISTINCT slots since v1.15.19, so the percentages are
right even without running this script. Running it additionally clears
the phantom rows from the history view.
- The `intake-slot-dedup` worker already collapses canonical-slot
duplicates at boot and on a daily cron; this script is the manual,
immediately-observable fallback and the only path that surfaces the
@@ -104,13 +104,9 @@ healthlog-tsx scripts/repair-intake-anomalies.ts --fix
as the duplicate losers). A medication without any derivable anchor
(PRN-only / unscheduled) is skipped — there is nothing to compare
against.
- **Affected compliance rollups are recomputed** for every touched
`(user, medication, day)`. The script executes the same DISTINCT-slot
aggregation SQL the shared helper runs (a verbatim twin of
`recomputeMedicationComplianceForDay` in
`src/lib/rollups/medication-compliance-rollups.ts` — kept inline so
the script needs no app imports), so the scheduled/taken counts and
the rate self-correct immediately.
- **No separate recompute is needed.** Every compliance reader folds the
rows that share one slot, so the scheduled/taken counts and the rate are
right as soon as the duplicates are tombstoned.
The script is idempotent: a second `--fix` run finds zero duplicate groups
and changes nothing. Exit code is 0 on success (including a clean
@@ -0,0 +1,32 @@
-- Drop `medication_compliance_rollups`.
--
-- The per-(user, medication, day) coverage ledger lost its last reader when
-- the dashboard tile and the compliance views moved to the schedule-anchored
-- computation; every dose write still upserted a row into it, a boot backfill
-- refolded it, and it travelled through the data wipe for nothing. The write
-- hook now refreshes only what still derives from a dose day (the supplement
-- nutrients), and the boot retires the backfill queue.
--
-- Nothing is lost: the table held counts derived from
-- `medication_intake_events`, which stay.
--
-- Lock wait. Dropping the table removes its two foreign keys, and removing a
-- foreign key's triggers takes ACCESS EXCLUSIVE on the referenced tables,
-- `users` and `medications`, for the instant the catalog changes. Waiting for
-- that lock behind a long reader would queue every login and every
-- medication read behind the migration, so the wait is bounded at five
-- seconds inside one `DO` block (one statement, so it commits or rolls back
-- whole; `migrate deploy` runs a file statement by statement). A lock
-- timeout changes nothing and the file is safe to apply again, which the
-- marker declares (docs/ops/deploy.md, "A migration that timed out on a
-- lock").
--
-- @retry-on-lock-timeout
DO $migration$
BEGIN
PERFORM set_config('lock_timeout', '5s', true);
DROP TABLE IF EXISTS "medication_compliance_rollups";
END
$migration$;
+16 -58
View File
@@ -118,61 +118,61 @@ model User {
environmentAqHistoryJson Json? @map("environment_aq_history_json")
// Codex OAuth (ChatGPT subscription)
codexAccessTokenEncrypted String? @map("codex_access_token_encrypted")
codexRefreshTokenEncrypted String? @map("codex_refresh_token_encrypted")
codexAccessTokenEncrypted String? @map("codex_access_token_encrypted")
codexRefreshTokenEncrypted String? @map("codex_refresh_token_encrypted")
/// @internal: dead mirror of the expiry inside the encrypted Codex creds —
/// the refresh path reads the decrypted blob's own `expiresAt`
/// (`src/lib/ai/provider.ts`), never this column. Kept (and kept written)
/// for wire/backup compatibility rather than dropped.
codexTokenExpiresAt DateTime? @map("codex_token_expires_at")
codexConnectedAt DateTime? @map("codex_connected_at")
codexConnectionStatus String @default("disconnected") @map("codex_connection_status")
codexTokenExpiresAt DateTime? @map("codex_token_expires_at")
codexConnectedAt DateTime? @map("codex_connected_at")
codexConnectionStatus String @default("disconnected") @map("codex_connection_status")
// Per-user opt-in to the operator's shared central Codex (ChatGPT
// subscription) connection stored on `AppSettings.adminCodex*`. OFF by
// default; only takes effect when the operator has actually connected the
// central Codex. Routing through it is external egress on a shared,
// train-by-default account bound by the operator's rate limits — consent-gated
// exactly like any other server-managed egress.
useCentralCodex Boolean @default(false) @map("use_central_codex")
insightsPrivacyMode String @default("aggregated") @map("insights_privacy_mode") // "aggregated" | "raw"
insightsCachedAt DateTime? @map("insights_cached_at")
useCentralCodex Boolean @default(false) @map("use_central_codex")
insightsPrivacyMode String @default("aggregated") @map("insights_privacy_mode") // "aggregated" | "raw"
insightsCachedAt DateTime? @map("insights_cached_at")
/// @internal: legacy plaintext cache of the AI insights payload, read only
/// as a fallback for rows the free-text encryption backfill has not reached
/// yet (v1.43.7). Every write since then goes to `insightsCachedTextEncrypted`
/// and leaves this NULL; the backfill seals and nulls it on the old rows.
/// DROP this column in a later release, once the backfill reports zero
/// remaining rows on every instance.
insightsCachedText String? @map("insights_cached_text")
insightsCachedText String? @map("insights_cached_text")
/// v1.43.7 — the cached AI insights payload (briefing, per-domain readings,
/// recommendations; JSON), AES-256-GCM binary layout under
/// `INSIGHTS_CACHED_TEXT_AAD`. Read and written only through
/// `src/lib/insights/cached-text-cipher.ts`.
insightsCachedTextEncrypted Bytes? @map("insights_cached_text_encrypted")
insightsCachedTextEncrypted Bytes? @map("insights_cached_text_encrypted")
// Language the cached text was generated in. The cache is one slot per
// user, so a reader in another language must not be served it: the
// snapshot briefing lift and the advisor read treat a tag that differs
// from the reader's locale as an empty cache and re-warm in the reader's.
// NULL on rows written before the tag existed (treated as matching; the
// next write tags it).
insightsCachedLocale String? @map("insights_cached_locale")
insightsCachedLocale String? @map("insights_cached_locale")
// v1.16.8 — SHA-256 fingerprint of the compacted feature snapshot the
// cached comprehensive text was generated from. Nightly / forced / lazy
// regenerations compare the fresh snapshot's hash against this and skip
// the provider call (refreshing only `insightsCachedAt`) when nothing
// the prompt sees has changed.
insightsSnapshotHash String? @map("insights_snapshot_hash")
insightsSnapshotHash String? @map("insights_snapshot_hash")
// v1.16.8 — last time a forced full warm's comprehensive step failed or
// timed out. The forced warm path backs off for an hour after a failure
// so a persistently broken provider chain cannot be re-driven into a
// full generation attempt by every page-open.
insightsWarmFailedAt DateTime? @map("insights_warm_failed_at")
insightsWarmFailedAt DateTime? @map("insights_warm_failed_at")
// v1.18.7 — UTC YYYY-MM-DD of the last daily-briefing phrasing re-roll.
// The content-hash gate keeps the FINDINGS byte-stable on unchanged data,
// but the daily-briefing PARAGRAPH is re-rolled once per calendar day at a
// higher, seedless temperature so the prose reads fresh day-over-day
// without re-paying for the findings. This stamps the day a re-roll ran so
// the gate stays at most one extra call per day.
insightsBriefingRerollDate String? @map("insights_briefing_reroll_date")
insightsBriefingRerollDate String? @map("insights_briefing_reroll_date")
// Morning-freshness marker: the user-local YYYY-MM-DD (their profile
// timezone) for which the event-driven morning refresh has run — i.e. last
// night's sleep has arrived AND the sleep-dependent insight/score cache was
@@ -181,13 +181,13 @@ model User {
// `morning-digest-refresh` job the sleep-arrival trigger enqueues; the
// string form makes the "same local morning" comparison timezone-honest
// without a second timestamp read.
morningDigestRefreshedOn String? @map("morning_digest_refreshed_on")
morningDigestRefreshedOn String? @map("morning_digest_refreshed_on")
// v1.4.36 W3 T3 — per-user AI Insights exclude-metrics list, mirrors
// `coachPrefsJson.excludeMetrics`. The /api/insights/generate route
// filters matching keys off `features` before JSON.stringify so the
// model never sees data the user opted out of. Default `[]` = legacy
// behaviour for every existing row.
insightsExcludeMetrics String[] @default([]) @map("insights_exclude_metrics")
insightsExcludeMetrics String[] @default([]) @map("insights_exclude_metrics")
// v1.8.5 — user-level injection-site deny-list. Sites listed here are
// never offered for ANY medication's injection-site picker and the
@@ -897,11 +897,6 @@ model User {
// read serves from this table instead of the unbounded
// MoodEntry.findMany walk.
moodEntryRollups MoodEntryRollup[]
// v1.4.39 W-MED — per-(user, medication, day) compliance rollup.
// Per-write hook on MedicationIntakeEvent mutations + the scheduler
// keep the row in sync so `/api/medications/intake?scope=compliance`
// and `computeUserHealthScoreFastPath` skip the per-event findMany.
medicationComplianceRollups MedicationComplianceRollup[]
// v1.4.40 SB-10 — AI consent receipts the iOS app POSTs when the user
// grants AI processing during onboarding. Persisted audit trail
// covers App-Store Guideline 5.1.2(i) ("data-collection consent must
@@ -2972,10 +2967,6 @@ model Medication {
/// roll-up (GI / METABOLIC / INJECTION_SITE / COGNITIVE /
/// GLP1_SPECIFIC) and 1-5 Likert severity per entry.
sideEffects MedicationSideEffect[]
/// v1.4.39 W-MED — per-(medication, day) compliance rollup. Mutations
/// on the linked MedicationIntakeEvent rows refresh the row inline;
/// the read paths consume this instead of the per-event findMany.
complianceRollups MedicationComplianceRollup[]
/// v1.16.3 — archived schedule eras (effective dating). Superseded
/// schedule states land here when a wholesale schedule replace changes
/// cadence-relevant fields; the band minter segments historical minting
@@ -5038,39 +5029,6 @@ model MoodEntryRollup {
@@map("mood_entry_rollups")
}
// ─── Medication Compliance Rollup (v1.4.39 W-MED) ─────────────
//
// Per-(userId, medicationId, day) compliance ledger. Replaces the
// unbounded MedicationIntakeEvent findMany on
// `/api/medications/intake?scope=compliance` and the equivalent
// fan-out inside `computeUserHealthScoreFastPath`.
//
// Day-key is anchored to the user's IANA timezone (TEXT, encoded as
// `YYYY-MM-DD`) so the read path can collapse a date range into a
// straight WHERE clause without re-deriving the boundary per row.
//
// Write path: every MedicationIntakeEvent create / update / delete
// (and the scheduler that mints fresh `scheduledFor` rows) fires
// `recomputeMedicationComplianceForDay` which upserts the
// `(user, medication, day)` row. Worker handles weekly / monthly
// re-fold off the hot path.
model MedicationComplianceRollup {
userId String @map("user_id")
medicationId String @map("medication_id")
day String // YYYY-MM-DD anchored to user.timezone
scheduled Int
taken Int
skipped Int
computedAt DateTime @default(now()) @map("computed_at") @db.Timestamptz(3)
user User @relation(fields: [userId], references: [id], onDelete: Cascade)
medication Medication @relation(fields: [medicationId], references: [id], onDelete: Cascade)
@@id([userId, medicationId, day])
@@index([userId, day(sort: Desc)])
@@map("medication_compliance_rollups")
}
// ─── Integration Status (v1.4.15) ─────────────────────────
//
// Per-(user, integration) row that captures the most recent sync attempt
+2 -2
View File
@@ -107,8 +107,8 @@ acknowledgement).
`audit_logs` rows (the text lives in the `details` JSON) instead.
- **Transient / regenerable:** idempotency_keys, rate_limits, data_backups,
import_jobs, host_metrics, feedback, recommendation_feedback, and all rollup
tables (measurement_rollups, mood_entry_rollups,
medication_compliance_rollups, strain_trimp_cache, cycle_predictions) — these
tables (measurement_rollups, mood_entry_rollups, strain_trimp_cache,
cycle_predictions) — these
rebuild on edge01.
- **No rows:** workouts / workout_routes / workout_samples (demo has none).
- **Instance / global:** app_settings; global catalogues (mood_tags,
+1 -3
View File
@@ -65,8 +65,7 @@
# - Ops / transient / regenerable: idempotency_keys, rate_limits,
# data_backups, import_jobs, host_metrics, feedback,
# recommendation_feedback, all rollup tables (measurement_rollups,
# mood_entry_rollups, medication_compliance_rollups, strain_trimp_cache,
# cycle_predictions) — these REGENERATE on edge01.
# mood_entry_rollups, strain_trimp_cache, cycle_predictions) — these REGENERATE on edge01.
# - workouts / workout_routes / workout_samples — demo has none.
# - app_settings — instance-level, not user data.
# - Global catalogues (mood_tags, cycle_symptoms) — pre-exist on edge01;
@@ -299,7 +298,6 @@ DELETE FROM audit_logs WHERE user_id = '$EDGE_DEMO_ID';
-- regenerable / transient tables that may carry FK rows blocking the user delete
DELETE FROM measurement_rollups WHERE user_id = '$EDGE_DEMO_ID';
DELETE FROM mood_entry_rollups WHERE user_id = '$EDGE_DEMO_ID';
DELETE FROM medication_compliance_rollups WHERE user_id = '$EDGE_DEMO_ID';
DELETE FROM strain_trimp_cache WHERE user_id = '$EDGE_DEMO_ID';
DELETE FROM cycle_predictions WHERE user_id = '$EDGE_DEMO_ID';
DELETE FROM measurements WHERE user_id = '$EDGE_DEMO_ID';
+6 -159
View File
@@ -50,14 +50,9 @@
* CREATE/UPDATE only with `--backfill-eras`. Medications that
* already have a revision are skipped (idempotent).
*
* After any mutation the affected `(user, medication, day)` compliance
* rollups are recomputed with the SAME slot-level DISTINCT aggregation the
* shared helper runs — see `recomputeComplianceDay` below, a verbatim
* SQL twin of `recomputeMedicationComplianceForDay` in
* `src/lib/rollups/medication-compliance-rollups.ts`. The rollup
* aggregates DISTINCT slots since v1.15.19, so a recompute alone already
* heals the counts; the tombstones additionally clear the phantom rows
* from the history view.
* The tombstones clear the phantom rows from the history view; every
* compliance reader folds rows sharing a slot, so the counts need no
* separate recompute.
*
* Default is a DRY-RUN that only reports. Idempotent: a second `--fix`
* run finds zero groups and changes nothing.
@@ -211,8 +206,8 @@ function pickWinner(rows: IntakeRow[]): IntakeRow {
// Timezone / day-key helpers — self-contained twins of the app helpers
// the previous revision imported. Reference implementations:
// - `userDayKey` / `DEFAULT_TIMEZONE`: `src/lib/tz/format.ts`
// - `tzOffsetMinutes` / `startOfDayUtcInTz` / the DST-safe day-after
// boundary: `src/lib/rollups/medication-compliance-rollups.ts`
// - `tzOffsetMinutes` / `startOfDayUtcInTz`: `startOfLocalDayKey` in
// `src/lib/tz/local-day.ts`
// ─────────────────────────────────────────────────────────────────────
const DEFAULT_TIMEZONE = "Europe/Berlin";
@@ -287,100 +282,6 @@ function startOfDayUtcInTz(dayKey: string, tz: string): Date {
return guess;
}
/**
* `[start, windowEnd)` UTC window of the local calendar day — the same
* DST-safe boundary derivation `recomputeMedicationComplianceForDay`
* runs (23 / 25-hour days fold correctly because the end anchors on
* the NEXT day's local midnight, not `start + 24 h`).
*/
function dayWindowUtc(dayKey: string, tz: string): { start: Date; end: Date } {
const start = startOfDayUtcInTz(dayKey, tz);
const fallbackEnd = new Date(start.getTime() + 86_400_000);
const [y, m, d] = dayKey.split("-").map(Number);
const probeNext = new Date(Date.UTC(y, m - 1, d, 12, 0, 0) + 86_400_000);
const nextDayKey = dayKeyForScheduledFor(probeNext, tz);
const dstSafeEnd = startOfDayUtcInTz(nextDayKey, tz);
return {
start,
end: dstSafeEnd.getTime() > start.getTime() ? dstSafeEnd : fallbackEnd,
};
}
/**
* Recompute one `(userId, medicationId, dayKey)` compliance-rollup row.
*
* VERBATIM SQL TWIN of `recomputeMedicationComplianceForDay` in
* `src/lib/rollups/medication-compliance-rollups.ts` — keep the two in
* sync when the rollup contract changes. v1.15.19 semantics: the inner
* CTE folds rows sharing one `scheduled_for` instant into one DISTINCT
* slot with taken-beats-skipped priority; the outer aggregate counts
* slots, not rows, so a duplicated slot can never inflate `scheduled`.
* The upsert re-aggregates inside the statement (atomic against
* concurrent app writers); the trailing DELETE clears the row when the
* day no longer holds any live event.
*/
async function recomputeComplianceDay(
client: Client,
userId: string,
medicationId: string,
dayKey: string,
tz: string | null | undefined,
): Promise<void> {
const { start, end } = dayWindowUtc(dayKey, safeTimezone(tz));
await client.query(
`
WITH slot AS (
SELECT
BOOL_OR("taken_at" IS NOT NULL AND NOT "skipped") AS slot_taken,
BOOL_OR("skipped") AS slot_skipped
FROM "medication_intake_events"
WHERE "user_id" = $1
AND "medication_id" = $2
AND "deleted_at" IS NULL
AND "scheduled_for" >= $4::timestamp
AND "scheduled_for" < $5::timestamp
GROUP BY "scheduled_for"
),
aggregate AS (
SELECT
COUNT(*)::int AS scheduled,
COALESCE(SUM(CASE WHEN "slot_taken" THEN 1 ELSE 0 END), 0)::int AS taken,
COALESCE(SUM(CASE WHEN NOT "slot_taken" AND "slot_skipped" THEN 1 ELSE 0 END), 0)::int AS skipped
FROM slot
)
INSERT INTO "medication_compliance_rollups"
("user_id", "medication_id", "day", "scheduled", "taken", "skipped", "computed_at")
SELECT $1, $2, $3, aggregate.scheduled, aggregate.taken, aggregate.skipped, NOW()
FROM aggregate
WHERE aggregate.scheduled > 0
ON CONFLICT ("user_id", "medication_id", "day") DO UPDATE
SET "scheduled" = EXCLUDED."scheduled",
"taken" = EXCLUDED."taken",
"skipped" = EXCLUDED."skipped",
"computed_at" = EXCLUDED."computed_at"
`,
[userId, medicationId, dayKey, utcParam(start), utcParam(end)],
);
await client.query(
`
DELETE FROM "medication_compliance_rollups"
WHERE "user_id" = $1
AND "medication_id" = $2
AND "day" = $3
AND NOT EXISTS (
SELECT 1
FROM "medication_intake_events"
WHERE "user_id" = $1
AND "medication_id" = $2
AND "deleted_at" IS NULL
AND "scheduled_for" >= $4::timestamp
AND "scheduled_for" < $5::timestamp
)
`,
[userId, medicationId, dayKey, utcParam(start), utcParam(end)],
);
}
interface Summary {
duplicateGroups: number;
rowsTombstoned: number;
@@ -394,7 +295,6 @@ interface Summary {
eraRevisionsCreated: number;
startsOnCandidates: number;
startsOnAdjusted: number;
rollupsRecomputed: number;
}
const HHMM_RE = /^([01]\d|2[0-3]):([0-5]\d)$/;
@@ -453,7 +353,6 @@ async function main(): Promise<void> {
eraRevisionsCreated: 0,
startsOnCandidates: 0,
startsOnAdjusted: 0,
rollupsRecomputed: 0,
};
// Optional per-user scope. The fragment is a fixed string with a
@@ -461,7 +360,7 @@ async function main(): Promise<void> {
const userFilter = opts.userId ? `AND e."user_id" = $1` : "";
const userParams = opts.userId ? [opts.userId] : [];
/** `userId -> timezone` cache for the rollup day-key derivation. */
/** `userId -> timezone` cache for the local-day derivations. */
const tzCache = new Map<string, string | null>();
async function userTimezone(userId: string): Promise<string | null> {
if (!tzCache.has(userId)) {
@@ -474,10 +373,6 @@ async function main(): Promise<void> {
return tzCache.get(userId) ?? null;
}
// `(userId, medicationId, dayKey)` tuples whose compliance rollup needs
// a recompute after the tombstones. Deduped so a busy day folds once.
const daysToRecompute = new Set<string>();
try {
// ─────────────────────────────────────────────────────────────────
// 1. Slot duplicates — exact (user, medication, scheduled_for)
@@ -568,12 +463,6 @@ async function main(): Promise<void> {
[losers.map((r) => r.id)],
);
summary.rowsTombstoned += res.rowCount ?? 0;
const tz = await userTimezone(group.user_id);
const dayKey = dayKeyForScheduledFor(group.scheduled_for, tz);
daysToRecompute.add(
`${group.user_id}|${group.medication_id}|${dayKey}`,
);
}
}
@@ -651,12 +540,6 @@ async function main(): Promise<void> {
console.log(
`${TAG} tombstoned ${res.rowCount ?? 0} implausible row(s) (operator opt-in)`,
);
for (const row of implausible) {
const tz = await userTimezone(row.user_id);
const dayKey = dayKeyForScheduledFor(row.scheduled_for, tz);
daysToRecompute.add(`${row.user_id}|${row.medication_id}|${dayKey}`);
}
}
// ─────────────────────────────────────────────────────────────────
@@ -822,12 +705,6 @@ async function main(): Promise<void> {
[staleRows.map((r) => r.id)],
);
summary.staleAnchorTombstoned = res.rowCount ?? 0;
for (const row of staleRows) {
const tz = await userTimezone(row.user_id);
const dayKey = dayKeyForScheduledFor(row.scheduled_for, tz);
daysToRecompute.add(`${row.user_id}|${row.medication_id}|${dayKey}`);
}
}
// ─────────────────────────────────────────────────────────────────
@@ -1057,35 +934,6 @@ async function main(): Promise<void> {
summary.startsOnAdjusted += res.rowCount ?? 0;
}
}
// ─────────────────────────────────────────────────────────────────
// 6. Recompute the affected compliance rollups so scheduled/taken
// counts self-correct. Same DISTINCT-slot SQL the shared helper
// runs (see `recomputeComplianceDay` above).
// ─────────────────────────────────────────────────────────────────
if (opts.fix && daysToRecompute.size > 0) {
console.log(
`\n${TAG} 6) recomputing ${daysToRecompute.size} compliance rollup day(s)`,
);
for (const key of daysToRecompute) {
const [userId, medicationId, dayKey] = key.split("|");
try {
await recomputeComplianceDay(
client,
userId,
medicationId,
dayKey,
await userTimezone(userId),
);
summary.rollupsRecomputed += 1;
} catch (err) {
console.error(
`${TAG} rollup recompute failed for ${key}: ` +
`${err instanceof Error ? err.message : String(err)}`,
);
}
}
}
} finally {
await client.end();
}
@@ -1112,7 +960,6 @@ async function main(): Promise<void> {
console.log(` era revisions created: ${summary.eraRevisionsCreated}`);
console.log(` startsOn candidates: ${summary.startsOnCandidates}`);
console.log(` startsOn adjusted: ${summary.startsOnAdjusted}`);
console.log(` rollup days recomputed: ${summary.rollupsRecomputed}`);
}
main().catch((err) => {
-1
View File
@@ -37,7 +37,6 @@ END $$;
-- ── Wipe per-user state ───────────────────────────────────────────
DELETE FROM medication_intake_events WHERE user_id IN (SELECT user_id FROM _demo);
DELETE FROM medication_compliance_rollups WHERE user_id IN (SELECT user_id FROM _demo);
DELETE FROM medication_schedules WHERE medication_id IN (
SELECT id FROM medications WHERE user_id IN (SELECT user_id FROM _demo)
);
-4
View File
@@ -30,10 +30,6 @@ const ALLOWED: Record<string, { hits: number; why: string }> = {
hits: 1,
why: "computed_at is timestamptz",
},
"src/lib/rollups/medication-compliance-rollups.ts": {
hits: 1,
why: "computed_at is timestamptz",
},
"src/lib/jobs/job-failures.ts": {
hits: 2,
why: "pg-boss tables are timestamptz",
+1 -1
View File
@@ -85,7 +85,7 @@ import { __resetAllCachesForTests } from "@/lib/cache/server-cache";
import { serialiseDocument } from "@/lib/documents/store";
import { resolveIntakeActionability } from "@/lib/medications/intake-actionable";
import { moodDateKey } from "@/lib/mood/date-key";
import { dayKeyForScheduledFor } from "@/lib/rollups/medication-compliance-rollups";
import { dayKeyForScheduledFor } from "@/lib/medications/dose-day-refresh";
import {
dateOnlyAtNoonUtc,
dayKeyAsUtcMidnight,
@@ -23,14 +23,6 @@ vi.mock("@/lib/db-compat", () => ({
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
}));
// v1.37.19 — the zone change enqueues a compliance re-fold.
const refoldMock = vi.hoisted(() =>
vi.fn(async () => ({ enqueued: true, error: null })),
);
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
enqueueUserMedicationComplianceBackfill: refoldMock,
}));
// The supplement nutrient rows are keyed by the same local day.
const nutrientMock = vi.hoisted(() => vi.fn(async () => ({ queued: true })));
vi.mock("@/lib/medications/ingredients/recompute-job", () => ({
@@ -110,11 +102,9 @@ describe("PUT /api/auth/me/timezone", () => {
expect(res.status).toBe(422);
});
// Watched red: with the re-fold enqueue removed from the route (the
// pre-v1.37.19 write), the changed-zone case fails — the compliance
// rollup's day keys were minted under the old zone and nothing ever
// re-bucketed them (the coverage probe compares counts, not keys).
it("enqueues a compliance re-fold when the zone actually changes", async () => {
// The supplement nutrient rows' day keys were minted under the old zone;
// without the enqueue nothing re-buckets them.
it("enqueues the nutrient recompute when the zone actually changes", async () => {
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
vi.mocked(prisma.user.findUnique).mockResolvedValue({
timezone: "Europe/Berlin",
@@ -123,11 +113,10 @@ describe("PUT /api/auth/me/timezone", () => {
const res = await PUT(mkReq({ timezone: "America/New_York" }));
expect(res.status).toBe(200);
expect(refoldMock).toHaveBeenCalledWith("user-1");
expect(nutrientMock).toHaveBeenCalledWith("user-1");
});
it("does not re-fold when the zone is unchanged", async () => {
it("does not recompute when the zone is unchanged", async () => {
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
vi.mocked(prisma.user.findUnique).mockResolvedValue({
timezone: "Europe/Berlin",
@@ -136,7 +125,6 @@ describe("PUT /api/auth/me/timezone", () => {
const res = await PUT(mkReq({ timezone: "Europe/Berlin" }));
expect(res.status).toBe(200);
expect(refoldMock).not.toHaveBeenCalled();
expect(nutrientMock).not.toHaveBeenCalled();
});
+5 -17
View File
@@ -30,7 +30,6 @@ import { annotate } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import { invalidateUserTimezone, isValidTimezone } from "@/lib/tz/resolver";
import { enqueueSupplementNutrientRecompute } from "@/lib/medications/ingredients/recompute-job";
import { enqueueUserMedicationComplianceBackfill } from "@/lib/rollups/medication-compliance-rollups";
export const PUT = apiHandler(async (request: NextRequest) => {
const { user } = await requireAuth();
@@ -62,24 +61,13 @@ export const PUT = apiHandler(async (request: NextRequest) => {
invalidateUserTimezone(user.id);
// v1.37.19 — the medication-compliance rollup keys its rows by
// the LOCAL day the dose fell on, minted under the timezone in force at
// write time. A zone change re-buckets those days, so the stored keys go
// stale (and the coverage probe compares counts, not keys — it cannot
// notice). Re-fold the trailing window through the ordinary backfill
// job; best-effort, the singleton key collapses rapid re-picks.
// The supplement nutrient rows are keyed by the LOCAL day the dose fell
// on, minted under the timezone in force at write time. A zone change
// re-buckets those days, so the account's rows are recomputed through
// the ordinary job; best-effort, the singleton key collapses rapid
// re-picks.
if ((before?.timezone ?? null) !== tz) {
const refold = await enqueueUserMedicationComplianceBackfill(user.id);
// The supplement nutrient rows are keyed by the same local day.
await enqueueSupplementNutrientRecompute(user.id);
annotate({
meta: {
timezone_compliance_refold_enqueued: refold.enqueued,
...(refold.error
? { timezone_compliance_refold_error: refold.error }
: {}),
},
});
}
await auditLog("user.timezone.update", {
@@ -38,8 +38,8 @@ vi.mock("@/lib/rate-limit", () => ({
vi.mock("@/lib/app-settings", () => ({
isApiGloballyEnabled: vi.fn(),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/medications/inventory/consumption", () => ({
consumeForIntake: vi.fn().mockResolvedValue(undefined),
+2 -2
View File
@@ -23,7 +23,7 @@ import {
unstableExternalIdShape,
} from "@/lib/validations/external-id";
import { isApiGloballyEnabled } from "@/lib/app-settings";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
import {
applyCanonicalSlotWrite,
resolveSlotForWriteByBand,
@@ -337,7 +337,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
// v1.4.39 — refresh the compliance rollup for the ingested
// event's user-day.
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: apiToken.userId,
medicationId: medication.id,
scheduledFor: event.scheduledFor,
@@ -51,9 +51,9 @@ vi.mock("@/lib/cache/invalidate", () => ({
// v1.16.1 — the schedule-replace branch recomputes the compliance
// rollups of the tombstoned slots; stub the rollup module so the route
// test never reaches `$executeRaw`.
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
vi.mock("@/lib/medications/dose-day-refresh", () => ({
dayKeyForScheduledFor: vi.fn(() => "2026-06-10"),
recomputeMedicationComplianceForDay: vi.fn().mockResolvedValue(undefined),
refreshDoseDay: vi.fn().mockResolvedValue(undefined),
}));
// v1.40 (#1024) — the course writer and reader run against their own
// integration suite; here they answer "no course, nothing refused".
@@ -99,8 +99,8 @@ import {
import { auditLog } from "@/lib/auth/audit";
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForDay,
} from "@/lib/rollups/medication-compliance-rollups";
refreshDoseDay,
} from "@/lib/medications/dose-day-refresh";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
@@ -297,7 +297,7 @@ describe("PUT /api/medications/[id] — schedule replace archives a revision (v1
vi.mocked(getMedicationCategories).mockResolvedValue({});
vi.mocked(auditLog).mockResolvedValue(undefined);
vi.mocked(dayKeyForScheduledFor).mockReturnValue("2026-06-10");
vi.mocked(recomputeMedicationComplianceForDay).mockResolvedValue(undefined);
vi.mocked(refreshDoseDay).mockResolvedValue(undefined);
vi.mocked(prisma.medicationSchedule.findMany).mockResolvedValue([
PREVIOUS_ROW,
] as never);
@@ -447,19 +447,19 @@ describe("PUT /api/medications/[id] — schedule replace migrates open slots", (
ROUTE_CTX,
);
expect(res.status).toBe(200);
expect(recomputeMedicationComplianceForDay).toHaveBeenCalledTimes(2);
expect(recomputeMedicationComplianceForDay).toHaveBeenCalledWith(
"user-1",
"m1",
"2026-06-10",
"Europe/Berlin",
);
expect(recomputeMedicationComplianceForDay).toHaveBeenCalledWith(
"user-1",
"m1",
"2026-06-11",
"Europe/Berlin",
);
expect(refreshDoseDay).toHaveBeenCalledTimes(2);
expect(refreshDoseDay).toHaveBeenCalledWith({
userId: "user-1",
medicationId: "m1",
dayKey: "2026-06-10",
tz: "Europe/Berlin",
});
expect(refreshDoseDay).toHaveBeenCalledWith({
userId: "user-1",
medicationId: "m1",
dayKey: "2026-06-11",
tz: "Europe/Berlin",
});
});
it("does NOT touch intake rows when the body carries no schedules array", async () => {
@@ -865,7 +865,7 @@ describe("PUT /api/medications/[id] — intake tracking switch (#1033)", () => {
vi.mocked(getMedicationCategories).mockResolvedValue({});
vi.mocked(auditLog).mockResolvedValue(undefined);
vi.mocked(dayKeyForScheduledFor).mockReturnValue("2026-06-10");
vi.mocked(recomputeMedicationComplianceForDay).mockResolvedValue(undefined);
vi.mocked(refreshDoseDay).mockResolvedValue(undefined);
vi.mocked(prisma.medicationSchedule.findMany).mockResolvedValue([
LIVE_ROW,
] as never);
@@ -1055,7 +1055,7 @@ describe("PUT /api/medications/[id] — shipped client on a record-only medicati
vi.mocked(getMedicationCategories).mockResolvedValue({});
vi.mocked(auditLog).mockResolvedValue(undefined);
vi.mocked(dayKeyForScheduledFor).mockReturnValue("2026-06-10");
vi.mocked(recomputeMedicationComplianceForDay).mockResolvedValue(undefined);
vi.mocked(refreshDoseDay).mockResolvedValue(undefined);
vi.mocked(prisma.medicationSchedule.findMany).mockResolvedValue([
STORED_ROW,
] as never);
@@ -37,8 +37,8 @@ vi.mock("@/lib/medications/inventory/consumption", () => ({
consumeForIntake: vi.fn().mockResolvedValue(undefined),
restoreForIntake: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
vi.mock("@/lib/db-compat", () => ({
@@ -19,7 +19,7 @@ import {
import { reconcileOneShotState } from "@/lib/medications/lifecycle";
import { invalidateUserMedications } from "@/lib/cache/invalidate";
import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
import { dayKeyForUserTz } from "@/lib/measurements/consolidation-tz";
import {
applyCanonicalSlotWrite,
@@ -392,14 +392,14 @@ export const PUT = apiHandler(
// scheduledFor day-keys when the dose moved across a day boundary
// (band re-attribution can move it as well as an explicit edit); if the
// slot is unchanged the second call collapses to a no-op upsert.
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: user.id,
medicationId: id,
scheduledFor: event.scheduledFor,
tz: user.timezone,
});
if (slotMoved) {
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: user.id,
medicationId: id,
scheduledFor: targetScheduledFor,
@@ -488,7 +488,7 @@ export const DELETE = apiHandler(
// v1.4.39 — refresh the rollup row for the day the deleted
// event sat on. When that day now holds zero events the helper
// drops the rollup row entirely.
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: user.id,
medicationId: id,
scheduledFor: event.scheduledFor,
@@ -78,8 +78,8 @@ vi.mock("@/lib/cache/invalidate", () => ({
invalidateUserMedications: vi.fn(),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
@@ -86,8 +86,8 @@ vi.mock("@/lib/cache/invalidate", () => ({
invalidateUserMedications: vi.fn(),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/notifications/medication-intake-sync", () => ({
@@ -46,12 +46,12 @@ vi.mock("@/lib/rate-limit", () => ({
rateLimitHeaders: vi.fn(() => ({ "X-RateLimit-Remaining": "0" })),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
vi.mock("@/lib/medications/dose-day-refresh", () => ({
dayKeyForScheduledFor: vi.fn((d: Date) => {
const iso = d.toISOString();
return iso.slice(0, 10);
}),
recomputeMedicationComplianceForDay: vi.fn().mockResolvedValue(undefined),
refreshDoseDay: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/medications/inventory/consumption", () => ({
@@ -98,8 +98,8 @@ import { assertMedicationOwnership } from "@/lib/medications/route-guards";
import { checkRateLimit } from "@/lib/rate-limit";
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForDay,
} from "@/lib/rollups/medication-compliance-rollups";
refreshDoseDay,
} from "@/lib/medications/dose-day-refresh";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
@@ -174,7 +174,7 @@ describe("POST /api/medications/[id]/intake/bulk-delete", () => {
const json = await res.json();
expect(json.data).toEqual({ deleted: 0 });
expect(prisma.medicationIntakeEvent.updateMany).not.toHaveBeenCalled();
expect(recomputeMedicationComplianceForDay).not.toHaveBeenCalled();
expect(refreshDoseDay).not.toHaveBeenCalled();
// (#22) — a no-op delete queues no cross-device sync fan-out.
expect(queueMedicationIntakeSync).not.toHaveBeenCalled();
});
@@ -201,7 +201,7 @@ describe("POST /api/medications/[id]/intake/bulk-delete", () => {
expect(dayKeyForScheduledFor).toHaveBeenCalledTimes(3);
// Three rows, two unique dayKeys → two recompute calls.
expect(recomputeMedicationComplianceForDay).toHaveBeenCalledTimes(2);
expect(refreshDoseDay).toHaveBeenCalledTimes(2);
// (#22) — the whole bulk delete queues exactly ONE cross-device
// sync fan-out, not one per removed row.
@@ -15,8 +15,8 @@ import { restoreForIntake } from "@/lib/medications/inventory/consumption";
import { assertMedicationOwnership } from "@/lib/medications/route-guards";
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForDay,
} from "@/lib/rollups/medication-compliance-rollups";
refreshDoseDay,
} from "@/lib/medications/dose-day-refresh";
import { invalidateUserMedications } from "@/lib/cache/invalidate";
import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
import { NextRequest } from "next/server";
@@ -159,26 +159,11 @@ export const POST = apiHandler(
},
});
// Best-effort rollup refresh — one trip per unique dayKey so a
// 14-row delete spanning two days closes in two SQL trips. The
// helper is idempotent and the catch swallows transient
// populator failures (same contract as the single-event delete).
// Best-effort dose-day refresh, one per unique dayKey, so a 14-row
// delete spanning two days refreshes two days (same contract as the
// single-event delete).
for (const dayKey of dayKeysToRefresh) {
try {
await recomputeMedicationComplianceForDay(user.id, id, dayKey, tz);
} catch (err) {
annotate({
action: {
name: "medications.intake.bulk-delete.recompute-failed",
entity_type: "medication",
entity_id: id,
},
meta: {
day_key: dayKey,
error: err instanceof Error ? err.message : String(err),
},
});
}
await refreshDoseDay({ userId: user.id, medicationId: id, dayKey, tz });
}
await auditLog("medication.intake.bulk_delete", {
@@ -2,9 +2,9 @@
* v1.5.5 — purge route fires the medication cache bundle.
*
* Pins the contract the detail-page Verlauf-löschen cascade depends
* on: dropping the rollup rows is necessary but not sufficient. The
* on: tombstoning the rows is necessary but not sufficient. The
* analytics + iOS today-tally + dashboard tiles all read off cached
* shapes that survive the rollup delete for their TTL; the bundle
* shapes that survive the tombstoning for their TTL; the bundle
* invalidation forces every downstream reader to converge on the
* post-purge counts in the same tick the user sees the toast.
*/
@@ -16,7 +16,6 @@ vi.mock("@/lib/db", () => ({
prisma: {
medication: { findUnique: vi.fn() },
medicationIntakeEvent: { updateMany: vi.fn() },
medicationComplianceRollup: { deleteMany: vi.fn() },
// v1.43 — no composition, so no supplement recompute is queued.
medicationIngredient: { findFirst: vi.fn().mockResolvedValue(null) },
},
@@ -87,9 +86,6 @@ beforeEach(() => {
vi.mocked(prisma.medicationIntakeEvent.updateMany).mockResolvedValue({
count: 7,
} as never);
vi.mocked(prisma.medicationComplianceRollup.deleteMany).mockResolvedValue({
count: 5,
} as never);
});
describe("DELETE /api/medications/[id]/intake/purge", () => {
@@ -17,8 +17,8 @@ type RouteParams = { params: Promise<{ id: string }> };
export const DELETE = apiHandler(
async (request: NextRequest, { params }: RouteParams) => {
// v1.37.0 — MANAGE. It tombstones rather than deletes (since the
// delegated-write release) and drops only rollups that recompute, so the
// history it clears is reconstructable from the rows themselves.
// delegated-write release), so the history it clears is reconstructable
// from the rows themselves.
const { user } = await requireRecordAuth("manage", "medications");
const { id } = await params;
@@ -60,13 +60,6 @@ export const DELETE = apiHandler(
data: { deletedAt: new Date(), syncVersion: { increment: 1 } },
});
// v1.4.39 — drop every rollup row for this medication so the
// next compliance read returns zero-filled buckets rather than the
// last-known scheduled / taken totals for now-deleted events.
await prisma.medicationComplianceRollup.deleteMany({
where: { medicationId: id, userId: user.id },
});
// v1.43 — the purged doses no longer count toward the micronutrients.
if (await medicationHasComposition(user.id, id)) {
await enqueueSupplementNutrientRecompute(user.id);
+2 -2
View File
@@ -36,7 +36,7 @@ import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake
import { dispatchMedicationIntakeWebClear } from "@/lib/notifications/web-push-clear";
import { countOutstandingDosesToday } from "@/lib/medications/outstanding-doses";
import { notifyDelegatedIntake } from "@/lib/notifications/delegated-intake";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
import {
applyCanonicalSlotWrite,
findPinConflict,
@@ -608,7 +608,7 @@ async function postIntake(request: NextRequest, { params }: RouteParams) {
// v1.4.39 — refresh the persistent compliance rollup for the
// affected day. The hook is best-effort; failures annotate but never
// block the user's POST response.
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: user.id,
medicationId: id,
scheduledFor: event.scheduledFor,
+15 -25
View File
@@ -48,8 +48,8 @@ import {
} from "@/lib/medications/scheduling/next-due";
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForDay,
} from "@/lib/rollups/medication-compliance-rollups";
refreshDoseDay,
} from "@/lib/medications/dose-day-refresh";
import { assertMedicationOwnership } from "@/lib/medications/route-guards";
import {
effectiveUnitsPerDose,
@@ -716,9 +716,9 @@ export const PUT = apiHandler(
// Tombstone today's and future open pending rows of this medication.
// Runs on a schedule replace (the old anchors are stale) and when intake
// tracking is switched off (nothing is due any more).
// The rollup days the tombstoned slots sat on, recomputed once the
// The dose days the tombstoned slots sat on, refreshed once the
// transaction below has committed the tombstones.
const staleRollupDays = new Set<string>();
const staleDoseDays = new Set<string>();
const tombstoneOpenSlots = async (
db: Prisma.TransactionClient,
): Promise<void> => {
@@ -768,7 +768,7 @@ export const PUT = apiHandler(
});
// Deduped via the user-tz day key.
for (const row of tombstoned) {
staleRollupDays.add(dayKeyForScheduledFor(row.scheduledFor, userTz));
staleDoseDays.add(dayKeyForScheduledFor(row.scheduledFor, userTz));
}
};
@@ -1085,26 +1085,16 @@ export const PUT = apiHandler(
throw err;
}
// Recompute each rollup day the tombstoned slots sat on, now that the
// tombstones are committed. Best-effort like every other rollup
// write-hook: a failure is annotated, never fails the edit.
for (const dayKey of staleRollupDays) {
try {
await recomputeMedicationComplianceForDay(
user.id,
id,
dayKey,
user.timezone || DEFAULT_TIMEZONE,
);
} catch (err) {
annotate({
meta: {
medication_compliance_rollup_failed: true,
medication_compliance_rollup_error:
err instanceof Error ? err.message : String(err),
},
});
}
// Refresh each dose day the tombstoned slots sat on, now that the
// tombstones are committed. Best-effort like every dose write-hook: a
// failure is reported, never fails the edit.
for (const dayKey of staleDoseDays) {
await refreshDoseDay({
userId: user.id,
medicationId: id,
dayKey,
tz: user.timezone || DEFAULT_TIMEZONE,
});
}
// Apply the v1.5.5 primary-schedule grace bridge after the
@@ -45,7 +45,6 @@ import {
toRevisionPayloadEntry,
type ScheduleRevisionEntry,
} from "@/lib/medications/scheduling/schedule-eras";
import { enqueueUserMedicationComplianceBackfill } from "@/lib/rollups/medication-compliance-rollups";
import { invalidateUserMedications } from "@/lib/cache/invalidate";
import type { Prisma } from "@/generated/prisma/client";
@@ -286,13 +285,10 @@ export const PATCH = apiHandler(
},
});
// The corrected era re-segments history; refresh the pre-aggregated
// compliance rollups asynchronously (best-effort).
// v1.16.9 — an era write re-segments the bands every cached payload
// (list next-due, compliance cells, dashboard tally) was built on;
// hard-evict so the next read reflects the new history immediately.
invalidateUserMedications(user.id, { evict: true });
await enqueueUserMedicationComplianceBackfill(user.id);
return apiSuccess({
id: outcome.revision.id,
@@ -404,13 +400,10 @@ export const DELETE = apiHandler(
meta: { revision_id: revisionId },
});
// History re-segments without the era; refresh the pre-aggregated
// compliance rollups asynchronously (best-effort).
// v1.16.9 — an era write re-segments the bands every cached payload
// (list next-due, compliance cells, dashboard tally) was built on;
// hard-evict so the next read reflects the new history immediately.
invalidateUserMedications(user.id, { evict: true });
await enqueueUserMedicationComplianceBackfill(user.id);
return apiSuccess({ deleted: true });
},
@@ -42,11 +42,6 @@ vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
vi.mock("@/lib/auth/audit", () => ({
auditLog: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
enqueueUserMedicationComplianceBackfill: vi
.fn()
.mockResolvedValue({ enqueued: true, error: null }),
}));
vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
vi.mock("@/lib/db-compat", () => ({
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
@@ -64,7 +59,6 @@ import { GET, POST } from "../route";
import { DELETE, PATCH } from "../[revisionId]/route";
import { prisma } from "@/lib/db";
import { getSession } from "@/lib/auth/session";
import { enqueueUserMedicationComplianceBackfill } from "@/lib/rollups/medication-compliance-rollups";
const SESSION_OK = {
session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) },
@@ -349,11 +343,6 @@ describe("POST /api/medications/[id]/schedule-revisions", () => {
expect(createArgs.data.payload[0].rrule).toBe("FREQ=DAILY");
expect(createArgs.data.payload[0].scheduleType).toBe("SCHEDULED");
// History re-segmented — the rollup backfill must be enqueued.
expect(enqueueUserMedicationComplianceBackfill).toHaveBeenCalledWith(
"user-1",
);
// Check-then-insert ran inside the transaction with the
// per-medication `FOR UPDATE` lock taken first.
expect(prisma.$transaction).toHaveBeenCalled();
@@ -432,9 +421,6 @@ describe("DELETE /api/medications/[id]/schedule-revisions/[revisionId]", () => {
where: { medicationId: "m1", supersededByRevisionId: "rev-m" },
data: { supersededByRevisionId: null },
});
expect(enqueueUserMedicationComplianceBackfill).toHaveBeenCalledWith(
"user-1",
);
});
});
@@ -533,9 +519,6 @@ describe("PATCH /api/medications/[id]/schedule-revisions/[revisionId]", () => {
expect(updateArgs.data.payload[0].rrule).toBe("FREQ=DAILY");
// In-place — no superseding row is minted.
expect(prisma.medicationScheduleRevision.create).not.toHaveBeenCalled();
expect(enqueueUserMedicationComplianceBackfill).toHaveBeenCalledWith(
"user-1",
);
// Check-then-write ran inside the locked transaction.
expect(prisma.$transaction).toHaveBeenCalled();
expect(prisma.$queryRaw).toHaveBeenCalled();
@@ -573,8 +556,5 @@ describe("PATCH /api/medications/[id]/schedule-revisions/[revisionId]", () => {
where: { id: "rev-t" },
data: { supersededByRevisionId: "rev-fix" },
});
expect(enqueueUserMedicationComplianceBackfill).toHaveBeenCalledWith(
"user-1",
);
});
});
@@ -40,7 +40,6 @@ import {
toRevisionPayloadEntry,
type ScheduleRevisionEntry,
} from "@/lib/medications/scheduling/schedule-eras";
import { enqueueUserMedicationComplianceBackfill } from "@/lib/rollups/medication-compliance-rollups";
import { invalidateUserMedications } from "@/lib/cache/invalidate";
import type { Prisma } from "@/generated/prisma/client";
@@ -282,14 +281,10 @@ export const POST = apiHandler(
},
});
// The new era re-segments history the compliance rollups already
// pre-aggregated; refresh them asynchronously. Best-effort like
// every other rollup write-hook.
// v1.16.9 — an era write re-segments the bands every cached payload
// (list next-due, compliance cells, dashboard tally) was built on;
// hard-evict so the next read reflects the new history immediately.
invalidateUserMedications(user.id, { evict: true });
await enqueueUserMedicationComplianceBackfill(user.id);
return apiSuccess(
{
@@ -34,12 +34,6 @@ vi.mock("@/lib/db", () => ({
// v1.15.18 — the band resolver loads the medication via findFirst.
findFirst: vi.fn(),
},
medicationComplianceRollup: {
findFirst: vi.fn(),
findMany: vi.fn(),
upsert: vi.fn(),
deleteMany: vi.fn(),
},
// v1.4.43 — audit-ledger breadcrumb for validation-failed paths.
auditLog: { create: vi.fn() },
$transaction: vi.fn(),
@@ -119,15 +113,6 @@ beforeEach(() => {
vi.mocked(prisma.medicationIntakeEvent.createMany).mockResolvedValue({
count: 0,
} as never);
// v1.4.39 — default the coverage probe to "uncovered" so the
// legacy compliance test still exercises the live-fallback branch
// and finds the legacy mocked intake events.
vi.mocked(prisma.medicationComplianceRollup.findFirst).mockResolvedValue(
null,
);
vi.mocked(prisma.medicationComplianceRollup.findMany).mockResolvedValue(
[] as never,
);
// v1.4.39 — the coverage probe is now a single `$queryRaw`
// aggregate returning `{ rolled_days, event_days }`. Default to
// "zero rollups, zero events" (covered/trivial-empty) so tests that
@@ -51,8 +51,8 @@ vi.mock("@/lib/medications/inventory/consumption", () => ({
consumeForIntake: vi.fn().mockResolvedValue(undefined),
restoreForIntake: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForDay: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseDay: vi.fn().mockResolvedValue(undefined),
dayKeyForScheduledFor: vi.fn().mockReturnValue("2026-01-01"),
}));
vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
+14 -29
View File
@@ -65,9 +65,9 @@ import {
} from "@/lib/integrations/healthkit-type-sync";
import { invalidateUserMedications } from "@/lib/cache/invalidate";
import {
recomputeMedicationComplianceForDay,
dayKeyForScheduledFor,
} from "@/lib/rollups/medication-compliance-rollups";
refreshDoseDay,
} from "@/lib/medications/dose-day-refresh";
import {
applyCanonicalSlotWrite,
findPinConflict,
@@ -365,10 +365,9 @@ async function postBulk(request: NextRequest): Promise<Response> {
let extraDosesUnscheduled = 0;
const skipped: Array<{ index: number; reason: string }> = [];
const unstableShapes: UnstableExternalIdShape[] = [];
// v1.4.39 — collect distinct `(medicationId, dayKey)` pairs
// touched by the batch so one rollup recompute fires per pair after
// all inserts complete. Per-row recompute would balloon a 500-entry
// batch into 500 sequential rollup hits.
// Distinct `(medicationId, dayKey)` pairs touched by the batch, so one
// dose-day refresh fires per pair after all inserts complete. A per-row
// refresh would balloon a 500-entry batch into 500 sequential refreshes.
const touchedDays = new Map<
string,
{ medicationId: string; dayKey: string }
@@ -949,30 +948,16 @@ async function postBulk(request: NextRequest): Promise<Response> {
invalidateUserMedications(user.id, { evict: true });
}
// v1.4.39 — refresh one rollup row per distinct
// `(medicationId, dayKey)` touched by the batch. Best-effort: failures
// are swallowed by the helper's `recomputeMedicationComplianceForEvent`
// wrapper, but bulk callers wrap recompute directly to keep the
// tight ingest loop unaffected by populator issues.
// Refresh the derived rows of each distinct `(medicationId, dayKey)` the
// batch touched. Best-effort: the refresh reports its own failures and
// never fails the ingest.
for (const { medicationId, dayKey } of touchedDays.values()) {
try {
await recomputeMedicationComplianceForDay(
user.id,
medicationId,
dayKey,
user.timezone,
);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
annotate({
meta: {
medication_compliance_rollup_bulk_failed: true,
medication_compliance_rollup_bulk_error: message,
medication_compliance_rollup_medication: medicationId,
medication_compliance_rollup_day: dayKey,
},
});
}
await refreshDoseDay({
userId: user.id,
medicationId,
dayKey,
tz: user.timezone,
});
}
// v1.17.1 (#22) — silent cross-device intake sync. ONE queued fan-out
+3 -3
View File
@@ -28,7 +28,7 @@ import { auditLog } from "@/lib/auth/audit";
import { DEFAULT_TIMEZONE } from "@/lib/tz/resolver";
import { cached, caches, type ServerCache } from "@/lib/cache/server-cache";
import { invalidateUserMedications } from "@/lib/cache/invalidate";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
import {
buildScheduleAnchoredComplianceBuckets,
type ScheduleAnchoredComplianceBucket,
@@ -573,7 +573,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
// the affected day so the next read after the cache miss returns
// the up-to-date `(scheduled, taken, skipped)` tuple. Best-effort:
// a populator failure annotates ops without blocking the response.
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: user.id,
medicationId: existing.medicationId,
scheduledFor: existing.scheduledFor,
@@ -584,7 +584,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
// source day self-corrects while the destination keeps its stale tuple.
const movedTo = updated?.scheduledFor;
if (movedTo && movedTo.getTime() !== existing.scheduledFor.getTime()) {
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: user.id,
medicationId: existing.medicationId,
scheduledFor: movedTo,
@@ -93,8 +93,8 @@ vi.mock("@/lib/cache/invalidate", () => ({
invalidateUserMedications: vi.fn(),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
import { POST, GET } from "../route";
-1
View File
@@ -75,7 +75,6 @@ export const WIPE_MODELS = [
"MedicationSideEffect",
"MedicationInventoryItem",
"MedicationIntakeImportJob",
"MedicationComplianceRollup",
// The account's own category names (v1.40, #1041). No child rows; the
// medications that name one go with their own entry.
"MedicationCategoryLabel",
@@ -44,7 +44,7 @@ vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
vi.mock("@/lib/rollups/mood-rollups", () => ({
recomputeUserMoodRollups: vi.fn(),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
vi.mock("@/lib/medications/dose-day-refresh", () => ({
recomputeUserMedicationCompliance: vi.fn(),
MEDICATION_COMPLIANCE_BACKFILL_DAYS: 30,
}));
@@ -66,7 +66,7 @@ vi.mock("@/lib/cache/invalidate", () => ({ invalidateUserData: vi.fn() }));
vi.mock("@/lib/rollups/mood-rollups", () => ({
recomputeUserMoodRollups: vi.fn(),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
vi.mock("@/lib/medications/dose-day-refresh", () => ({
recomputeUserMedicationCompliance: vi.fn(),
MEDICATION_COMPLIANCE_BACKFILL_DAYS: 30,
}));
-2
View File
@@ -655,8 +655,6 @@ export const COVERAGE_PENDING: Readonly<Record<string, string>> = {
export const DERIVED_MODELS: Readonly<Record<string, string>> = {
MeasurementRollup:
"A per-day, per-metric reconstruction of `Measurement`. The boot-time rollup backfill rebuilds it for an account that has none.",
MedicationComplianceRollup:
"Recomputed from `MedicationIntakeEvent` against the schedule by the compliance analytics tier.",
MoodEntryRollup:
"A per-day reconstruction of `MoodEntry`, rebuilt by the same rollup tier that rebuilds the measurement one.",
MoodPrediction:
-31
View File
@@ -80,10 +80,6 @@ import {
type BackupSummary,
} from "@/lib/validations/backup-summary";
import { recomputeUserMoodRollups } from "@/lib/rollups/mood-rollups";
import {
recomputeUserMedicationCompliance,
MEDICATION_COMPLIANCE_BACKFILL_DAYS,
} from "@/lib/rollups/medication-compliance-rollups";
import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups";
import { restoreCycleData } from "@/lib/cycle/backup";
import {
@@ -3058,33 +3054,6 @@ export async function restoreBackup(
}
}
// v1.4.39 — re-fold the medication-compliance rollup tier
// from the restored intake events. The medication delete inside the
// transaction already cascaded the existing rollup partition for
// this owner via the FK, so the fold mints fresh rows. Boot-time
// backfill is the safety net if this best-effort call fails.
if (payload.intakeEvents.length > 0) {
try {
const restoreUser = await prisma.user.findUnique({
where: { id: ownerId },
select: { timezone: true },
});
await recomputeUserMedicationCompliance(
ownerId,
MEDICATION_COMPLIANCE_BACKFILL_DAYS,
restoreUser?.timezone ?? null,
);
} catch (err) {
annotate({
meta: {
medication_compliance_rollup_restore_failed: true,
medication_compliance_rollup_restore_error:
err instanceof Error ? err.message : String(err),
},
});
}
}
const summary = {
...summarizeBackup(payload),
measurements: measurementsRestored,
@@ -20,8 +20,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
vi.mock("@/lib/medications/inventory/consumption", () => ({
consumeForIntake: vi.fn(async () => undefined),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn(async () => undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn(async () => undefined),
dayKeyForScheduledFor: (at: Date, tz: string) =>
at.toLocaleDateString("sv-SE", { timeZone: tz }),
}));
@@ -36,7 +36,7 @@ vi.mock("@/lib/medications/lifecycle", () => ({
}));
import { consumeForIntake } from "@/lib/medications/inventory/consumption";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
import {
INTAKE_AUTO_TAKE_CRON,
@@ -590,7 +590,7 @@ function row(partial: Partial<FakeRow> & { id: string; scheduledFor: Date }) {
describe("runIntakeAutoTakePass — over a fake client", () => {
beforeEach(() => {
vi.mocked(consumeForIntake).mockClear();
vi.mocked(recomputeMedicationComplianceForEvent).mockClear();
vi.mocked(refreshDoseSlot).mockClear();
vi.mocked(queueMedicationIntakeSync).mockClear();
});
@@ -3,7 +3,6 @@ import { describe, expect, it } from "vitest";
import {
MEDICATION_IMPORT_SKIP_DETAIL_LIMIT,
MEDICATION_INTAKE_IMPORT_CHUNK_SIZE,
MEDICATION_INTAKE_IMPORT_ROLLUP_CHUNK_SIZE,
_setMedicationIntakeImportPrismaForTests,
advanceMedicationImportProgress,
appendMedicationImportSkipDetails,
@@ -48,21 +47,18 @@ describe("medication intake import worker bounds", () => {
processed: 100,
imported: 98,
skippedByReason: { duplicate_in_file: 2 },
touchedDays: ["2026-07-20", "2026-07-21"],
});
const second = advanceMedicationImportProgress(first, {
from: 100,
processed: 100,
imported: 97,
skippedByReason: { duplicate_in_file: 1, already_recorded: 2 },
touchedDays: ["2026-07-21", "2026-07-22"],
});
const replay = advanceMedicationImportProgress(second, {
from: 0,
processed: 100,
imported: 98,
skippedByReason: { duplicate_in_file: 2 },
touchedDays: ["2026-07-20"],
});
expect(first.processed).toBe(100);
@@ -71,7 +67,7 @@ describe("medication intake import worker bounds", () => {
total: 1_000,
imported: 195,
skippedByReason: { duplicate_in_file: 3, already_recorded: 2 },
touchedDays: ["2026-07-20", "2026-07-21", "2026-07-22"],
touchedDays: [],
rollupProcessed: 0,
});
expect(replay).toBe(second);
@@ -89,7 +85,11 @@ describe("medication intake import worker bounds", () => {
expect(failure.length).toBeLessThanOrEqual(1_000);
});
it("checkpoints bounded rollup transactions across many touched days", async () => {
it("finishes a job admitted with pending rollup days without recomputing them", async () => {
// A job admitted before the compliance rollup was dropped carries the
// days it touched and how many of them were recomputed. It now finishes
// on its next pass, writes nothing per day, and keeps its stored
// progress shape.
const touchedDays = Array.from({ length: 1_000 }, (_, index) =>
new Date(Date.UTC(2020, 0, index + 1)).toISOString().slice(0, 10),
);
@@ -99,32 +99,27 @@ describe("medication intake import worker bounds", () => {
imported: 0,
skippedByReason: {},
touchedDays,
rollupProcessed: 40,
};
let status = "running";
const executeRawCounts: number[] = [];
let executeRawCount = 0;
let transactions = 0;
let auditCount = 0;
let transactionNumber = 0;
let rejectRollup = true;
const client = {
$transaction: async (
operation: (tx: Record<string, unknown>) => Promise<unknown>,
) => {
transactionNumber += 1;
let executeRawCount = 0;
transactions += 1;
const tx = {
$queryRaw: async () => [{ id: "job-many-days" }],
$queryRaw: async () => [{ id: "job-legacy-days" }],
$executeRaw: async () => {
executeRawCount += 1;
if (rejectRollup && transactionNumber === 3) {
rejectRollup = false;
throw new Error("rollup transaction failed");
}
return 1;
},
medicationIntakeImportJob: {
findUnique: async () => ({
id: "job-many-days",
id: "job-legacy-days",
recordUserId: "user-1",
actorUserId: "user-1",
medicationId: "medication-1",
@@ -132,7 +127,6 @@ describe("medication intake import worker bounds", () => {
payload: { entries: [] },
progress: persistedProgress,
startedAt: new Date(),
recordUser: { timezone: "UTC" },
}),
update: async (args: {
data: { progress?: unknown; status?: string };
@@ -151,43 +145,25 @@ describe("medication intake import worker bounds", () => {
},
},
};
try {
return await operation(tx);
} finally {
executeRawCounts.push(executeRawCount);
}
return operation(tx);
},
};
_setMedicationIntakeImportPrismaForTests(client as never);
try {
await expect(
processMedicationIntakeImportJob("job-many-days"),
).rejects.toThrow("rollup transaction failed");
expect(persistedProgress).toMatchObject({
rollupProcessed: MEDICATION_INTAKE_IMPORT_ROLLUP_CHUNK_SIZE * 2,
});
expect(status).toBe("running");
expect(auditCount).toBe(0);
await processMedicationIntakeImportJob("job-many-days");
await processMedicationIntakeImportJob("job-legacy-days");
} finally {
_setMedicationIntakeImportPrismaForTests(null);
}
expect(executeRawCounts).toHaveLength(
Math.ceil(
touchedDays.length / MEDICATION_INTAKE_IMPORT_ROLLUP_CHUNK_SIZE,
) + 1,
);
expect(Math.max(...executeRawCounts)).toBeLessThanOrEqual(
MEDICATION_INTAKE_IMPORT_ROLLUP_CHUNK_SIZE * 2,
);
expect(status).toBe("done");
expect(persistedProgress).toMatchObject({
rollupProcessed: touchedDays.length,
});
expect(transactions).toBe(1);
expect(executeRawCount).toBe(0);
expect(auditCount).toBe(1);
expect(persistedProgress).toMatchObject({
touchedDays,
rollupProcessed: 40,
});
});
});
@@ -229,14 +205,12 @@ describe("medication import skip reporting (#650)", () => {
processed: 100,
imported: 90,
skippedByReason: { duplicate_in_file: 10 },
touchedDays: [],
});
const second = advanceMedicationImportProgress(first, {
from: 100,
processed: 100,
imported: 95,
skippedByReason: { already_recorded: 5 },
touchedDays: [],
});
expect(second.skippedByReason).toEqual({
@@ -99,9 +99,6 @@ vi.mock("@/lib/notifications/dispatcher", () => ({
}));
vi.mock("@/lib/crypto", () => ({ decrypt: vi.fn(() => "token") }));
vi.mock("@/lib/telegram", () => ({ deleteMessage: vi.fn() }));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/notifications/reminder-dedup", () => ({
claimNotificationEvent: vi.fn().mockResolvedValue(true),
medicationReminderDedupKey: vi.fn(() => "dedup-key"),
@@ -124,9 +124,6 @@ vi.mock("@/lib/notifications/dispatcher", () => ({
}));
vi.mock("@/lib/crypto", () => ({ decrypt: vi.fn(() => "token") }));
vi.mock("@/lib/telegram", () => ({ deleteMessage: vi.fn() }));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
}));
import { handleReminderCheck } from "@/lib/jobs/reminder/medication-reminder-check";
import { dispatchNotification } from "@/lib/notifications/dispatcher";
@@ -60,7 +60,6 @@ describe("rollup registrar queue policies", () => {
it.each([
["ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"],
["MOOD_ROLLUP_FULL_BACKFILL_QUEUE", "exclusive"],
["MEDICATION_COMPLIANCE_BACKFILL_QUEUE", "exclusive"],
["STEP_CONSOLIDATION_QUEUE", "exclusive"],
["STEP_CONSOLIDATION_REPAIR_QUEUE", "exclusive"],
["CUMULATIVE_PR_REDERIVE_QUEUE", "exclusive"],
+3 -3
View File
@@ -85,8 +85,8 @@ import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake
import { isP2002 } from "@/lib/prisma-errors";
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForEvent,
} from "@/lib/rollups/medication-compliance-rollups";
refreshDoseSlot,
} from "@/lib/medications/dose-day-refresh";
import { DEFAULT_TIMEZONE, validTimezoneOr } from "@/lib/tz/format";
import { getEvent } from "@/lib/logging/context";
@@ -375,7 +375,7 @@ async function autoTakeOne(
const dayKey = dayKeyForScheduledFor(slot.at, userTz);
if (days.has(dayKey)) continue;
days.add(dayKey);
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: medication.userId,
medicationId: medication.id,
scheduledFor: slot.at,
+7 -120
View File
@@ -9,15 +9,10 @@ import { jobBudget } from "@/lib/jobs/job-budget";
import { enqueueSupplementNutrientRecompute } from "@/lib/medications/ingredients/recompute-job";
import { consumeImportedIntakesBatch } from "@/lib/medications/inventory/consumption";
import { queueMedicationIntakeSync } from "@/lib/notifications/medication-intake-sync";
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForDay,
} from "@/lib/rollups/medication-compliance-rollups";
export const MEDICATION_INTAKE_IMPORT_QUEUE = "medication-intake-import";
export const MEDICATION_INTAKE_IMPORT_CONCURRENCY = 2;
export const MEDICATION_INTAKE_IMPORT_CHUNK_SIZE = 100;
export const MEDICATION_INTAKE_IMPORT_ROLLUP_CHUNK_SIZE = 20;
export const MEDICATION_INTAKE_IMPORT_STALE_AFTER_MS = 15 * 60_000;
export const MEDICATION_INTAKE_IMPORT_SEND_OPTIONS = {
@@ -136,6 +131,11 @@ export interface MedicationImportProgress {
skipDetails?: MedicationImportSkipDetail[];
/** Details beyond the hard cap, counted without retaining their contents. */
skippedDetailsOmitted?: number;
/**
* Wire fields of the dropped compliance rollup's recompute phase. A job
* admitted now carries `[]` and `0` and the worker never advances them;
* they stay on the published progress shape for clients that decode it.
*/
touchedDays: string[];
rollupProcessed: number;
}
@@ -218,7 +218,6 @@ export interface MedicationImportChunkResult {
imported: number;
skippedByReason: MedicationImportSkipCounts;
skipDetails?: MedicationImportSkipDetail[];
touchedDays: string[];
}
let workerPrismaSingleton: PrismaClient | null = null;
@@ -251,15 +250,6 @@ export function advanceMedicationImportProgress(
previous.total,
previous.processed + Math.max(0, Math.trunc(chunk.processed)),
);
const touchedDays = [...previous.touchedDays];
const seenDays = new Set(touchedDays);
for (const day of chunk.touchedDays) {
if (!seenDays.has(day)) {
seenDays.add(day);
touchedDays.push(day);
}
}
const skippedByReason: MedicationImportSkipCounts = {
...previous.skippedByReason,
};
@@ -293,7 +283,7 @@ export function advanceMedicationImportProgress(
}
: {}),
rollupProcessed: previous.rollupProcessed,
touchedDays,
touchedDays: previous.touchedDays,
};
}
@@ -503,67 +493,6 @@ interface ProcessChunkOutcome {
userId: string | null;
}
/**
* A day whose compliance rollup an import invalidated, as one string.
*
* A job can now span a whole regimen, so a bare day key no longer identifies
* what to recompute — two medications touched on the same day are two rollup
* rows. The medication is folded in ahead of a single space, which occurs in
* neither a cuid nor a `YYYY-MM-DD` key. Not a NUL byte: this value is persisted
* in the progress blob and Postgres rejects a NUL byte inside a jsonb string.
*/
const TOUCHED_DAY_SEPARATOR = " ";
function encodeTouchedDay(medicationId: string, dayKey: string): string {
return `${medicationId}${TOUCHED_DAY_SEPARATOR}${dayKey}`;
}
/**
* Split a stored touched-day back into its parts.
*
* A value with no separator was written before a job could span more than one
* medication, so it is a bare day key belonging to the job's own medication.
* Returns null when neither reading yields a medication — a per-medication job
* that lost its medication cannot be resolved and the day is left alone rather
* than recomputed against a guess.
*/
function decodeTouchedDay(
value: string,
fallbackMedicationId: string | null,
): { medicationId: string; dayKey: string } | null {
const separator = value.indexOf(TOUCHED_DAY_SEPARATOR);
if (separator === -1) {
if (!fallbackMedicationId) return null;
return { medicationId: fallbackMedicationId, dayKey: value };
}
const medicationId = value.slice(0, separator);
const dayKey = value.slice(separator + 1);
if (medicationId.length === 0 || dayKey.length === 0) return null;
return { medicationId, dayKey };
}
async function recomputeTouchedDays(
client: Pick<PrismaClient, "$executeRaw">,
userId: string,
jobMedicationId: string | null,
timezone: string,
days: readonly string[],
): Promise<void> {
await Promise.all(
days.map((day) => {
const decoded = decodeTouchedDay(day, jobMedicationId);
if (!decoded) return Promise.resolve();
return recomputeMedicationComplianceForDay(
userId,
decoded.medicationId,
decoded.dayKey,
timezone,
client,
);
}),
);
}
async function processNextChunk(
client: PrismaClient,
jobId: string,
@@ -587,7 +516,6 @@ async function processNextChunk(
const row = await tx.medicationIntakeImportJob.findUnique({
where: { id: jobId },
include: { recordUser: { select: { timezone: true } } },
});
if (!row || row.status === "done" || row.status === "failed") {
return {
@@ -601,40 +529,6 @@ async function processNextChunk(
const payload = parsePayload(row.payload);
const progress = parseProgress(row.progress, payload.entries.length);
if (progress.processed >= progress.total) {
const rollupDays = progress.touchedDays.slice(
progress.rollupProcessed,
progress.rollupProcessed + MEDICATION_INTAKE_IMPORT_ROLLUP_CHUNK_SIZE,
);
await recomputeTouchedDays(
tx,
row.recordUserId,
row.medicationId,
row.recordUser.timezone,
rollupDays,
);
const finalizedProgress: MedicationImportProgress = {
...progress,
rollupProcessed: progress.rollupProcessed + rollupDays.length,
};
if (finalizedProgress.rollupProcessed < progress.touchedDays.length) {
const heartbeatAt = new Date();
await tx.medicationIntakeImportJob.update({
where: { id: row.id },
data: {
status: "running",
progress: toJson(finalizedProgress),
heartbeatAt,
startedAt: row.startedAt ?? heartbeatAt,
failureReason: null,
},
});
return {
terminal: false,
finalized: false,
result: null,
userId: row.recordUserId,
};
}
const result: MedicationImportResult = {
imported: progress.imported,
skipped: totalMedicationImportSkips(progress.skippedByReason),
@@ -665,7 +559,7 @@ async function processNextChunk(
data: {
status: "done",
result: toJson(result),
progress: toJson(finalizedProgress),
progress: toJson(progress),
failureReason: null,
heartbeatAt: completedAt,
completedAt,
@@ -782,12 +676,6 @@ async function processNextChunk(
});
}
const touchedDays = created.map((event) =>
encodeTouchedDay(
event.medicationId,
dayKeyForScheduledFor(event.scheduledFor, row.recordUser.timezone),
),
);
// Two different facts, counted apart. The Map collapse above is entries
// of this submission that resolve to the same instant; `skipDuplicates`
// then drops whatever the ledger already holds for that instant, from an
@@ -803,7 +691,6 @@ async function processNextChunk(
already_recorded: uniqueEntries.length - created.length,
},
skipDetails: [...duplicateDetails, ...alreadyRecordedDetails],
touchedDays,
});
const heartbeatAt = new Date();
await tx.medicationIntakeImportJob.update({
-3
View File
@@ -203,9 +203,6 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
"mood-rollup-full-backfill": short(
"One account's mood entries, a few a day.",
),
"medication-compliance-full-backfill": short(
"One account's 90-day compliance window.",
),
"step-consolidation": short(
"One account's legacy step rows, one transaction per day.",
),
@@ -8,7 +8,6 @@ import { type Job } from "pg-boss";
import type { Locale } from "@/lib/i18n/config";
import { resolveJobLocale } from "@/lib/i18n/job-locale";
import { recordError, recordReminderCheck } from "@/lib/jobs/worker-status";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { decrypt } from "@/lib/crypto";
import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
import { withBackgroundEvent } from "@/lib/logging/background";
@@ -708,15 +707,6 @@ export async function handleReminderCheck(
missedDosesMinted++;
evt.addMeta("missed_dose", `${med.name}:${slotTime}`);
// v1.4.39 — refresh the compliance rollup so the
// per-day `scheduled` count increments before any read.
await recomputeMedicationComplianceForEvent({
userId: med.user.id,
medicationId: med.id,
scheduledFor: slotScheduledFor,
tz: med.user.timezone,
});
}
}
+9 -82
View File
@@ -38,13 +38,6 @@ import {
type MoodRollupFullBackfillPayload,
type MoodRollupRecomputePayload,
} from "@/lib/rollups/mood-rollups";
import {
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY,
recomputeUserMedicationCompliance,
enqueueBootTimeMedicationComplianceBackfill,
type MedicationComplianceBackfillPayload,
} from "@/lib/rollups/medication-compliance-rollups";
import {
SUPPLEMENT_NUTRIENT_RECOMPUTE_QUEUE,
SUPPLEMENT_NUTRIENT_RECOMPUTE_CONCURRENCY,
@@ -114,6 +107,7 @@ import {
createAndSchedule,
createAndWork,
cronIsTheRetry,
retireQueues,
type QueuePolicyTable,
type ScheduleEntry,
} from "./registrar-shared";
@@ -154,10 +148,6 @@ const allQueues = [
MOOD_ROLLUP_RECOMPUTE_QUEUE,
// v1.4.39 — boot-time fold queue for the mood rollup tier.
MOOD_ROLLUP_FULL_BACKFILL_QUEUE,
// v1.4.39 — boot-time fold for the medication-compliance
// rollup tier. Discovery enqueues one job per user with intake
// events but no rollup coverage; idempotent across reboots.
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
// v1.5.6 — boot-time legacy step consolidation. Discovery enqueues
// one job per user still holding live pre-v1.5.0 granular step rows.
STEP_CONSOLIDATION_QUEUE,
@@ -248,11 +238,6 @@ const queuePolicies: QueuePolicyTable = {
reason:
"Per-user mood backfill; boot-discovery driven and self-converging on rollup coverage.",
},
[MEDICATION_COMPLIANCE_BACKFILL_QUEUE]: {
policy: "exclusive",
reason:
"Per-user compliance backfill; boot-discovery driven and self-converging on rollup coverage.",
},
[STEP_CONSOLIDATION_QUEUE]: {
policy: "exclusive",
reason:
@@ -289,6 +274,13 @@ const queuePolicies: QueuePolicyTable = {
// empty key and buy nothing.
};
/**
* The boot backfill of the per-day medication-compliance rollup, whose table
* had no reader left and was dropped. Its queue, and any job still waiting in
* it, is removed on an instance that still has it.
*/
const RETIRED_ROLLUP_QUEUES = ["medication-compliance-full-backfill"] as const;
/**
* Register every rollup / consolidation queue. Returns the queue names created
* (for the boot-level aggregate assertion).
@@ -297,6 +289,7 @@ export async function registerRollupQueues(
boss: PgBoss,
): Promise<readonly string[]> {
await createAndSchedule(boss, allQueues, schedules, queuePolicies);
await retireQueues(boss, RETIRED_ROLLUP_QUEUES);
// v1.5.0 — persistent measurement rollup worker. Folds the
// WEEK / MONTH / YEAR buckets that the write-path hooks enqueue;
@@ -722,44 +715,6 @@ export async function registerRollupQueues(
},
);
// v1.4.39 — medication-compliance boot-backfill worker. The
// discovery helper below sends one job per user with intake events
// but zero rollup coverage; this handler folds the trailing 90-day
// window per account. Concurrency-1 so the populator never crowds
// the request pool.
await createAndWork<MedicationComplianceBackfillPayload>(
boss,
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
{ localConcurrency: MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY },
async (jobs) => {
let rowsUpsertedTotal = 0;
for (const job of jobs) {
const { userId } = job.data;
try {
const { rowsUpserted, durationMs } =
await recomputeUserMedicationCompliance(userId);
rowsUpsertedTotal += rowsUpserted;
workerLog(
"info",
`[medication-compliance-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
);
} catch (err) {
recordError();
workerLog(
"error",
`[medication-compliance-backfill] user=${userId} failed`,
err,
);
throw err;
}
}
return jobDone({
users: jobs.length,
rows_upserted: rowsUpsertedTotal,
});
},
);
// v1.4.37 — nightly drain worker. Walks every user × cumulative
// type and folds per-sample APPLE_HEALTH rows older than the cutoff
// into one `stats:…` row per calendar day. Idempotent — a second run
@@ -1138,34 +1093,6 @@ export async function enqueueRollupBootDiscovery(): Promise<void> {
);
}
// v1.4.39 — medication compliance rollup tier. Mirrors the v1.4.35.1
// pattern: one job per user with intake events but no rollup coverage.
// Idempotent across reboots and singleton-keyed inside pg-boss.
try {
// Stage 4.
const { enqueued, skipped, error } =
await enqueueBootTimeMedicationComplianceBackfill(
BOOT_BACKFILL_STAGGER_SECONDS * 4,
);
if (error) {
workerLog(
"error",
`[medication-compliance-backfill] boot discovery failed: ${error}`,
);
} else {
workerLog(
"info",
`[medication-compliance-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
);
}
} catch (err) {
workerLog(
"error",
"[medication-compliance-backfill] boot discovery threw an unexpected error",
err,
);
}
// v1.28.37 — step-consolidation repair. Finds every account still holding
// a resurrectable tombstoned GOOGLE_HEALTH / FITBIT daily-total step row
// (swept by the pre-fix consolidation) and enqueues one repair job per
-1
View File
@@ -72,7 +72,6 @@ const subsystemSurface = {
"rollup-full-backfill": { audience: "system" },
"mood-rollup-recompute": { audience: "system" },
"mood-rollup-full-backfill": { audience: "system" },
"medication-compliance-full-backfill": { audience: "system" },
"step-consolidation": { audience: "system" },
"step-consolidation-repair": { audience: "system" },
"cumulative-pr-rederive": { audience: "system" },
@@ -35,8 +35,8 @@ vi.mock("@/lib/db", () => ({
vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() }));
vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
import { dedupeUserIntakeSlots } from "../intake-slot-dedup";
@@ -27,8 +27,8 @@ vi.mock("@/lib/db", () => ({
vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() }));
vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/cache/invalidate", () => ({
invalidateUserMedications: vi.fn(),
@@ -25,13 +25,13 @@ vi.mock("@/lib/db", () => ({
vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() }));
vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
import { dedupeUserIntakeSlots } from "../intake-slot-dedup";
import { prisma } from "@/lib/db";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
const TZ = "Europe/Berlin";
@@ -156,7 +156,7 @@ describe("dedupeUserIntakeSlots — per-slot error isolation", () => {
// succeeded → exactly one normalised row counted.
expect(summary.rowsNormalised).toBe(1);
// Both canonical days recomputed despite the collision.
expect(recomputeMedicationComplianceForEvent).toHaveBeenCalledTimes(2);
expect(refreshDoseSlot).toHaveBeenCalledTimes(2);
});
it("isolates a non-P2002 error per slot without dead-lettering the job", async () => {
@@ -45,8 +45,8 @@ vi.mock("@/lib/db", () => ({
vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() }));
vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
vi.mock("@/lib/medications/dose-day-refresh", () => ({
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
}));
import { dedupeUserIntakeSlots } from "../intake-slot-dedup";
+62
View File
@@ -0,0 +1,62 @@
/**
* The write-path hook every dose mutation funnels through: after a dose of a
* medication was taken, skipped, corrected, deleted, undone or recorded
* automatically, the rows derived from that dose's local day are recomputed.
*
* Today that is one derived row family, the micronutrients a taken
* supplement dose adds (`refreshSupplementNutrientsForDose`). The hook used to
* upsert a per-day compliance rollup as well; that table had no reader left
* and was dropped, so this is what remains of it, under a name that says what
* it does.
*
* Best-effort, like every derived row: a failure is reported by the refresh
* itself and never fails the dose write.
*/
import { refreshSupplementNutrientsForDose } from "@/lib/medications/ingredients/recompute";
import { DEFAULT_TIMEZONE, isValidTimezone, userDayKey } from "@/lib/tz/format";
function safeTimezone(tz: string | null | undefined): string {
return tz && isValidTimezone(tz) ? tz : DEFAULT_TIMEZONE;
}
/**
* The account-local day (`YYYY-MM-DD`) a dose slot belongs to: the local day
* of its `scheduledFor` in the account's timezone, falling back to the server
* default for a missing or unknown zone.
*/
export function dayKeyForScheduledFor(
scheduledFor: Date,
tz: string | null | undefined,
): string {
return userDayKey(scheduledFor, safeTimezone(tz));
}
/** Refresh the derived rows of one medication's local day. */
export async function refreshDoseDay(input: {
userId: string;
medicationId: string;
dayKey: string;
tz: string | null | undefined;
}): Promise<void> {
await refreshSupplementNutrientsForDose({
userId: input.userId,
medicationId: input.medicationId,
dayKey: input.dayKey,
timezone: safeTimezone(input.tz),
});
}
/** Refresh the derived rows of the local day a dose slot sits on. */
export async function refreshDoseSlot(input: {
userId: string;
medicationId: string;
scheduledFor: Date;
tz: string | null | undefined;
}): Promise<void> {
await refreshDoseDay({
userId: input.userId,
medicationId: input.medicationId,
dayKey: dayKeyForScheduledFor(input.scheduledFor, input.tz),
tz: input.tz,
});
}
+2 -2
View File
@@ -59,7 +59,7 @@ import type {
WorkerMedicationRow,
WorkerScheduleRow,
} from "@/lib/medications/scheduling/worker-helpers";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
export const INTAKE_SLOT_DEDUP_QUEUE = "intake-slot-dedup";
@@ -462,7 +462,7 @@ export async function dedupeUserIntakeSlots(
const sep = key.indexOf("|");
const medicationId = key.slice(0, sep);
const instantIso = key.slice(sep + 1);
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId,
medicationId,
scheduledFor: new Date(instantIso),
@@ -18,9 +18,6 @@
* 2. Reads existing rows in the today-window.
* 3. Inserts the missing rows (`skipDuplicates: true` for the
* `(userId, medicationId, scheduledFor, source)` unique index).
* 4. Fires one compliance-rollup recompute per distinct
* `(medicationId, dayKey)` so the rollup matches the new
* `scheduled` count before the next read.
*
* The caller owns its own typed re-read of `MedicationIntakeEvent`
* afterwards — the two routes return different projections of those
@@ -35,14 +32,12 @@ import {
scheduleEmitsInWindow,
} from "@/lib/medications/scheduling/worker-helpers";
import { localHmAsUtc } from "@/lib/tz/local-day";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { TRACKED_INTAKE_WHERE } from "@/lib/medications/intake-tracking";
import {
isBeforeLiveEra,
LIVE_ERA_REVISION_ARGS,
liveEraStart,
} from "@/lib/medications/scheduling/live-era";
import { dayKeyForScheduledFor } from "@/lib/rollups/medication-compliance-rollups";
export interface ProjectTodayIntakesResult {
projected: number;
@@ -224,36 +219,5 @@ export async function projectTodayIntakesAndRecompute(input: {
skipDuplicates: true,
});
// Close the v1.4.39.4 compliance-rollup hook gap: a bulk projection
// mints fresh `(medicationId, scheduledFor)` rows in PENDING state;
// without this hook the rollup row stays at its previous
// `scheduled` count, which inflates compliance % until the user
// actually logs against the new row. Coalesce by `(medicationId,
// dayKey)` so we fire one recompute per distinct day-tuple instead
// of one per row.
//
// `Promise.allSettled` keeps the best-effort contract explicit at
// the call site: the helper swallows internally today, but a
// future refactor that lets a throw escape would otherwise turn
// the parent POST into a 5xx.
const seenDayKeys = new Set<string>();
const recomputeJobs: Array<Promise<void>> = [];
for (const m of missing) {
// The compliance rollup's own day for the slot: two slots on different
// local days that share a UTC date must each get their recompute.
const key = `${m.medicationId}|${dayKeyForScheduledFor(m.scheduledFor, userTz)}`;
if (seenDayKeys.has(key)) continue;
seenDayKeys.add(key);
recomputeJobs.push(
recomputeMedicationComplianceForEvent({
userId,
medicationId: m.medicationId,
scheduledFor: m.scheduledFor,
tz: userTz,
}),
);
}
await Promise.allSettled(recomputeJobs);
return { projected: projected.length, backfilled: missing.length };
}
@@ -1355,13 +1355,15 @@ export const medicationImportProgress = z
touchedDays: z
.array(z.string())
.describe(
"Local calendar days (YYYY-MM-DD) touched so far, driving the chunked rollup recompute behind the write.",
"Deprecated and inert: always `[]` on a job admitted from v1.43.8, when the per-day compliance rollup it drove was removed. A job admitted earlier keeps the days it recorded. Kept for clients that decode it; do not read it.",
),
rollupProcessed: z
.number()
.int()
.min(0)
.describe("Of `touchedDays`, how many have had their rollup recomputed."),
.describe(
"Deprecated and inert: always `0` on a job admitted from v1.43.8. Kept for clients that decode it; do not read it.",
),
})
.meta({
id: "MedicationImportProgress",
+1 -1
View File
@@ -2326,7 +2326,7 @@ export const profilePaths: NonNullable<ZodOpenApiObject["paths"]> = {
summary: "Set the account timezone",
description:
"Lifted out of the profile patch so the picker can send one field without round-tripping the rest. Accepts any zone the running engine's `Intl` knows; anything else is 422 rather than stored, so the resolver never has to defend against a poisoned column. Audit-logged with the old and new zone.\n\n" +
"A change is not only a display preference. The medication-compliance rollup keys its rows by the LOCAL day a dose fell on, minted under whatever zone was in force at write time, so moving the zone re-buckets that history and the stored keys go stale — and the coverage probe compares counts rather than keys, so it cannot notice. The route therefore enqueues a re-fold of the trailing window whenever the zone actually changes. Best-effort and de-duplicated, so rapid re-picks collapse into one.",
"A change is not only a display preference. The supplement nutrient rows key on the LOCAL day a dose fell on, minted under whatever zone was in force at write time, so moving the zone re-buckets them. The route therefore enqueues a recompute of the account's supplement nutrients whenever the zone actually changes. Best-effort and de-duplicated, so rapid re-picks collapse into one.",
requestBody: {
required: true,
content: { "application/json": { schema: timezonePutRequest } },
@@ -1,403 +0,0 @@
/**
* v1.4.39 — unit tests for the medication-compliance rollup tier.
*
* `prisma` is mocked at the module boundary so the test can pin the
* intake-event fan-out + the rollup upsert / read shape without
* standing up a real Postgres. Integration coverage (real Postgres +
* the FK cascade behaviour) lives next to the route test suite when
* the W-MED phase grows them; this file proves the helpers'
* contract.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
const mocks = vi.hoisted(() => ({
intakeFindMany: vi.fn(),
rollupUpsert: vi.fn(),
rollupDeleteMany: vi.fn(),
rollupFindMany: vi.fn(),
rollupFindFirst: vi.fn(),
queryRaw: vi.fn(),
executeRaw: vi.fn(),
bossSend: vi.fn(),
getGlobalBossMock: vi.fn(),
}));
vi.mock("@/lib/db", () => ({
prisma: {
medicationIntakeEvent: {
findMany: mocks.intakeFindMany,
},
medicationComplianceRollup: {
upsert: mocks.rollupUpsert,
deleteMany: mocks.rollupDeleteMany,
findMany: mocks.rollupFindMany,
findFirst: mocks.rollupFindFirst,
},
$queryRaw: mocks.queryRaw,
$executeRaw: mocks.executeRaw,
},
}));
vi.mock("@/lib/jobs/boss-instance", () => ({
getGlobalBoss: () => mocks.getGlobalBossMock(),
}));
vi.mock("@/lib/logging/context", () => ({
annotate: vi.fn(),
}));
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForDay,
recomputeMedicationComplianceForEvent,
recomputeUserMedicationCompliance,
enqueueBootTimeMedicationComplianceBackfill,
enqueueUserMedicationComplianceBackfill,
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
} from "../medication-compliance-rollups";
const {
intakeFindMany,
rollupUpsert,
rollupDeleteMany,
rollupFindMany,
rollupFindFirst,
queryRaw,
executeRaw,
bossSend,
getGlobalBossMock,
} = mocks;
beforeEach(() => {
vi.resetAllMocks();
intakeFindMany.mockResolvedValue([]);
rollupUpsert.mockResolvedValue({});
rollupDeleteMany.mockResolvedValue({ count: 0 });
rollupFindMany.mockResolvedValue([]);
rollupFindFirst.mockResolvedValue(null);
queryRaw.mockResolvedValue([]);
executeRaw.mockResolvedValue(0);
});
afterEach(() => {
vi.useRealTimers();
});
describe("dayKeyForScheduledFor", () => {
it("anchors the dayKey on the user's tz wall clock", () => {
// 2026-05-17T23:30:00Z — Berlin local is 2026-05-18 01:30 (CEST).
const ts = new Date("2026-05-17T23:30:00.000Z");
expect(dayKeyForScheduledFor(ts, "Europe/Berlin")).toBe("2026-05-18");
// Los Angeles is UTC-7 (PDT) — still 2026-05-17 16:30 local.
expect(dayKeyForScheduledFor(ts, "America/Los_Angeles")).toBe("2026-05-17");
});
it("falls back to the server default when tz is null", () => {
const ts = new Date("2026-05-17T22:00:00.000Z");
expect(dayKeyForScheduledFor(ts, null)).toBe(
dayKeyForScheduledFor(ts, "Europe/Berlin"),
);
});
});
describe("recomputeMedicationComplianceForDay", () => {
/**
* Helper: pull the bind-variable values out of a Prisma tagged-template
* call so the test can inspect the user-id / day / window bounds the
* SQL was rendered with. Prisma's `$executeRaw` tag passes the
* template strings as the first arg and the substituted values as the
* remaining positional args.
*/
function bindValues(call: unknown[]): unknown[] {
return call.slice(1);
}
it("runs an atomic INSERT … ON CONFLICT upsert with the (user, medication, day) tuple", async () => {
// QA F-H-02 (v1.4.39): the recompute is now a single SQL statement
// that re-aggregates inside the upsert; the JS pre-fold from the
// previous tier is gone so the test inspects the bind values
// instead of the intake fold.
await recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
);
// Two statements fire: the INSERT … ON CONFLICT upsert and the
// companion DELETE for the "all events removed" branch.
expect(executeRaw).toHaveBeenCalledTimes(2);
const insertBinds = bindValues(executeRaw.mock.calls[0]);
expect(insertBinds).toContain("user-1");
expect(insertBinds).toContain("med-1");
expect(insertBinds).toContain("2026-05-18");
const deleteBinds = bindValues(executeRaw.mock.calls[1]);
expect(deleteBinds).toContain("user-1");
expect(deleteBinds).toContain("med-1");
expect(deleteBinds).toContain("2026-05-18");
});
it("aggregates at SLOT level so cross-source duplicate rows cannot inflate scheduled", async () => {
// v1.15.19 — the partial unique index carries `source`, so a pending
// REMINDER row and a taken API row can share one `scheduled_for`
// instant. A row-level COUNT(*) counted that slot twice. Pin that the
// statement groups by `scheduled_for` and folds each slot with
// taken-beats-skipped priority (BOOL_OR) before counting. Semantics
// (two live rows, one slot, different source → scheduled=1, taken=1)
// are proven on real Postgres in
// `tests/integration/medication-compliance-rollup-slot.test.ts`.
await recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
);
const insertSql = (executeRaw.mock.calls[0][0] as readonly string[]).join(
"?",
);
expect(insertSql).toMatch(/GROUP BY\s+"scheduled_for"/);
expect(insertSql).toMatch(
/BOOL_OR\("taken_at" IS NOT NULL AND NOT "skipped"\)/,
);
expect(insertSql).toMatch(/BOOL_OR\("skipped"\)/);
// Slot-level taken beats skipped: a slot is `skipped` only when no row
// in it is taken.
expect(insertSql).toMatch(/NOT "slot_taken" AND "slot_skipped"/);
// Scheduled counts slots (the grouped CTE), not raw event rows.
expect(insertSql).toMatch(/FROM slot/);
});
it("emits a paired DELETE so a fully-cleared day removes its row", async () => {
// The DELETE fires unconditionally and gates on NOT EXISTS — when
// the day still has events the predicate matches zero rows. The
// test pins that the DELETE statement was issued so a future
// refactor that drops the second statement breaks here.
await recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
);
expect(executeRaw).toHaveBeenCalledTimes(2);
});
it("is idempotent across repeated invocations", async () => {
await recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
);
await recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
);
expect(executeRaw).toHaveBeenCalledTimes(4);
expect(bindValues(executeRaw.mock.calls[0])).toEqual(
bindValues(executeRaw.mock.calls[2]),
);
expect(bindValues(executeRaw.mock.calls[1])).toEqual(
bindValues(executeRaw.mock.calls[3]),
);
});
it("binds a tz-anchored UTC window into the upsert", async () => {
await recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
);
const insertBinds = bindValues(executeRaw.mock.calls[0]);
const insertDates = insertBinds.filter((b): b is Date => b instanceof Date);
// Berlin 2026-05-18 00:00 → UTC 2026-05-17 22:00 (CEST is UTC+2).
expect(insertDates[0].toISOString()).toBe("2026-05-17T22:00:00.000Z");
// Window closes 24h later — Berlin 2026-05-19 00:00 → UTC 2026-05-18 22:00.
expect(insertDates[1].toISOString()).toBe("2026-05-18T22:00:00.000Z");
});
it("buckets the same UTC instant on different days for Berlin vs LA", async () => {
await recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
);
const berlinInsert = bindValues(executeRaw.mock.calls[0]).filter(
(b): b is Date => b instanceof Date,
);
executeRaw.mockClear();
await recomputeMedicationComplianceForDay(
"user-2",
"med-2",
"2026-05-17",
"America/Los_Angeles",
);
const laInsert = bindValues(executeRaw.mock.calls[0]).filter(
(b): b is Date => b instanceof Date,
);
// Berlin 2026-05-18 starts at UTC 2026-05-17 22:00.
expect(berlinInsert[0].toISOString()).toBe("2026-05-17T22:00:00.000Z");
// LA 2026-05-17 starts at UTC 2026-05-17 07:00 (PDT is UTC-7).
expect(laInsert[0].toISOString()).toBe("2026-05-17T07:00:00.000Z");
});
it("commits the strictest aggregate under concurrent recompute calls", async () => {
// QA F-H-02 race pin: two concurrent recomputes for the same
// (user, medication, day) must both go through the atomic SQL
// path. The earlier `findMany`-then-`upsert` pattern could
// interleave A-SELECT → B-SELECT → B-UPSERT (correct) → A-UPSERT
// (stale N-1); the atomic upsert serialises on the row lock so
// each statement re-aggregates a snapshot taken after the prior
// commit. The mock can't reproduce real Postgres concurrency, but
// the test pins that BOTH callers issue the atomic statement (not
// the legacy read-aggregate-then-upsert) so a regression to the
// racy pattern would fail this assertion.
await Promise.all([
recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
),
recomputeMedicationComplianceForDay(
"user-1",
"med-1",
"2026-05-18",
"Europe/Berlin",
),
]);
// 2 callers × 2 statements (INSERT + DELETE) each.
expect(executeRaw).toHaveBeenCalledTimes(4);
// No legacy intake-findMany was issued — the atomic path replaces it.
expect(intakeFindMany).not.toHaveBeenCalled();
// No legacy rollupUpsert / rollupDeleteMany either.
expect(rollupUpsert).not.toHaveBeenCalled();
expect(rollupDeleteMany).not.toHaveBeenCalled();
});
});
describe("recomputeMedicationComplianceForEvent", () => {
it("derives the dayKey from scheduledFor + tz then dispatches", async () => {
// 2026-05-17T23:30:00Z → Berlin day 2026-05-18.
await recomputeMedicationComplianceForEvent({
userId: "user-1",
medicationId: "med-1",
scheduledFor: new Date("2026-05-17T23:30:00.000Z"),
tz: "Europe/Berlin",
});
// The atomic upsert binds the dayKey as one of the positional
// arguments; assert "2026-05-18" reached the SQL layer.
const insertBinds = executeRaw.mock.calls[0].slice(1);
expect(insertBinds).toContain("2026-05-18");
});
it("swallows recompute errors so the parent write never blocks", async () => {
executeRaw.mockRejectedValueOnce(new Error("DB melted"));
await expect(
recomputeMedicationComplianceForEvent({
userId: "user-1",
medicationId: "med-1",
scheduledFor: new Date("2026-05-17T12:00:00.000Z"),
tz: "Europe/Berlin",
}),
).resolves.toBeUndefined();
});
});
describe("recomputeUserMedicationCompliance", () => {
it("upserts one rollup per (medication, day) pair returned by discovery", async () => {
queryRaw.mockResolvedValue([
{ medication_id: "med-1", day: "2026-05-17" },
{ medication_id: "med-1", day: "2026-05-18" },
{ medication_id: "med-2", day: "2026-05-18" },
]);
const result = await recomputeUserMedicationCompliance(
"user-1",
30,
"Europe/Berlin",
);
expect(result.rowsUpserted).toBe(3);
// Each (medication, day) pair issues two atomic SQL statements:
// the INSERT … ON CONFLICT upsert and the companion DELETE that
// gates on `NOT EXISTS`.
expect(executeRaw).toHaveBeenCalledTimes(6);
});
});
describe("enqueueBootTimeMedicationComplianceBackfill", () => {
it("is a no-op when no boss is attached", async () => {
getGlobalBossMock.mockReturnValue(null);
const result = await enqueueBootTimeMedicationComplianceBackfill();
expect(result).toEqual({ enqueued: 0, skipped: 0, error: null });
expect(queryRaw).not.toHaveBeenCalled();
});
it("enqueues one job per uncovered user", async () => {
getGlobalBossMock.mockReturnValue({ send: bossSend });
queryRaw.mockResolvedValue([{ id: "user-1" }, { id: "user-2" }]);
bossSend.mockResolvedValueOnce("job-1");
bossSend.mockResolvedValueOnce(null); // coalesced
const result = await enqueueBootTimeMedicationComplianceBackfill();
expect(result.enqueued).toBe(1);
expect(result.skipped).toBe(1);
expect(result.error).toBeNull();
expect(bossSend).toHaveBeenCalledWith(
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
expect.objectContaining({ userId: "user-1" }),
expect.objectContaining({
singletonKey: "medication-compliance-boot-backfill|user-1",
}),
);
});
});
describe("enqueueUserMedicationComplianceBackfill", () => {
it("is a no-op when no boss is attached", async () => {
getGlobalBossMock.mockReturnValue(null);
const result = await enqueueUserMedicationComplianceBackfill("user-1");
expect(result).toEqual({ enqueued: false, error: null });
// Critically: no `$queryRaw` cluster-wide LEFT JOIN scan.
expect(queryRaw).not.toHaveBeenCalled();
});
it("enqueues exactly one job for the caller's user and never touches the cluster-wide LEFT JOIN", async () => {
// QA F-SEC-M-01 (v1.4.39): pre-fix the request-path coverage-miss
// path called `enqueueBootTimeMedicationComplianceBackfill` which
// issued a cluster-wide `LEFT JOIN medication_intake_events ×
// medication_compliance_rollups` on every hit. The user-scoped
// helper sends one targeted job and never runs the discovery scan.
getGlobalBossMock.mockReturnValue({ send: bossSend });
bossSend.mockResolvedValueOnce("job-1");
const result = await enqueueUserMedicationComplianceBackfill("user-1");
expect(result).toEqual({ enqueued: true, error: null });
expect(queryRaw).not.toHaveBeenCalled();
expect(bossSend).toHaveBeenCalledTimes(1);
expect(bossSend).toHaveBeenCalledWith(
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
expect.objectContaining({ userId: "user-1" }),
expect.objectContaining({
singletonKey: "medication-compliance-boot-backfill|user-1",
}),
);
});
it("reports a coalesced enqueue as `enqueued:false`", async () => {
getGlobalBossMock.mockReturnValue({ send: bossSend });
bossSend.mockResolvedValueOnce(null);
const result = await enqueueUserMedicationComplianceBackfill("user-2");
expect(result).toEqual({ enqueued: false, error: null });
});
});
@@ -1,438 +0,0 @@
/**
* v1.4.39 — persistent medication-compliance rollup tier.
*
* Replaces the unbounded `MedicationIntakeEvent.findMany` walk on
* `/api/medications/intake?scope=compliance` with a ledger keyed
* `(userId, medicationId, day)` where `day` is `YYYY-MM-DD` anchored
* to the user's IANA timezone.
*
* Write hooks fire from every `MedicationIntakeEvent` mutation site
* (create / update / delete / bulk import) plus the reminder-worker's
* mint path that creates `takenAt:null` rows on RED-phase reminders.
* The dashboard tile it was built for reads the schedule-anchored
* compliance (`src/lib/analytics/schedule-anchored-compliance.ts`) since
* v1.26, and the per-day recompute below is now what keeps a day's
* supplement nutrients in step with its doses; the table itself has no
* reader left.
*
* v1.15.18 DESIGN DECISION — this tier stays a RAW per-day COVERAGE count,
* NOT the band-attributed compliance %. It aggregates, across ALL of a
* user's medications, "of the intake rows whose `scheduled_for` lands on
* this day, how many carry a non-null `taken_at`" — a coarse daily
* cross-medication coverage strip for the dashboard mini-tiles. The
* unified band-membership attribution (the `tallyComplianceFromLedger`
* keystone) is per-medication and per-cadence: it depends on the slot
* bands of each schedule and the user timezone, which cannot be expressed
* in a single cross-medication SQL `SUM(CASE …)`. Re-attributing every
* intake against its cadence's bands before the tally would mean running
* the band minter per `(medication, day)` inside the rollup, which defeats
* the rollup's whole purpose (bounded SQL, no per-row Node walk). The
* detail page + the medication card already read the band-attributed % via
* `calculateCompliance` / `complianceChips`; this tier deliberately remains
* the cheaper coverage signal and is documented as such so the two are not
* confused. The `scheduled` / `taken` column names reflect this: a day's
* `taken` is "slots with an intake logged taken", not "doses inside their
* on-time band".
*
* v1.15.19 — the count is SLOT-level, not row-level. The partial unique
* index on `(user, medication, scheduled_for, source)` lets two live rows
* share one slot instant when their `source` differs (e.g. a pending
* REMINDER row plus a taken API row for the same dose); a row-level
* `COUNT(*)` double-counted that slot as two scheduled doses. The
* aggregate now groups by `scheduled_for` first: `scheduled` counts
* distinct slot instants, and each slot folds its rows with
* taken-beats-skipped priority (`BOOL_OR` of taken, then skipped only when
* no row in the slot is taken). This neutralises cross-source duplicate
* rows retroactively — any recompute over pre-existing duplicates yields
* the deduplicated counts without a data migration.
*
* Audit anchor: `.planning/round-v1438-perf-analysis.md` §2.5 + §5 P4.
*/
import type { PrismaClient } from "@/generated/prisma/client";
import { prisma } from "@/lib/db";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { refreshSupplementNutrientsForDose } from "@/lib/medications/ingredients/recompute";
import { annotate } from "@/lib/logging/context";
import { DEFAULT_TIMEZONE, isValidTimezone, userDayKey } from "@/lib/tz/format";
import { localDayWindow, startOfLocalDayKey } from "@/lib/tz/local-day";
/**
* pg-boss queue name for the boot-time medication compliance backfill.
* Mirrors the `rollup-full-backfill` v1.4.35.1 pattern: discovery is
* idempotent across reboots and the worker upserts every
* `(user, medication, day)` row for the trailing window.
*/
export const MEDICATION_COMPLIANCE_BACKFILL_QUEUE =
"medication-compliance-full-backfill";
/** Worker concurrency cap — backfill stays serial to spare the pool. */
export const MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY = 1;
/** Trailing window the boot backfill folds per uncovered user. */
export const MEDICATION_COMPLIANCE_BACKFILL_DAYS = 90;
/** Payload `boss.send` carries onto the backfill queue. */
export interface MedicationComplianceBackfillPayload {
userId: string;
/** Wall-clock kick-off for debugging only. */
enqueuedAt: string;
}
/**
* Resolve a safe tz string — fall back to the server default when the
* supplied value is missing or not an IANA zone the host recognises.
*/
function safeTimezone(tz: string | null | undefined): string {
if (tz && isValidTimezone(tz)) return tz;
return DEFAULT_TIMEZONE;
}
/**
* Convert a `scheduledFor` UTC instant to the user's `YYYY-MM-DD`
* day-key. Re-exposed so the write-hook caller can derive the same
* `dayKey` the rollup row is anchored on.
*/
export function dayKeyForScheduledFor(
scheduledFor: Date,
tz: string | null | undefined,
): string {
return userDayKey(scheduledFor, safeTimezone(tz));
}
type MedicationComplianceRollupClient = Pick<PrismaClient, "$executeRaw">;
/**
* Recompute the rollup row for one `(userId, medicationId, dayKey)`
* tuple. Reads every `MedicationIntakeEvent` whose `scheduledFor` lands
* in `[startOfDayInTz, startOfNextDayInTz)` then upserts the
* `medication_compliance_rollups` row.
*
* The hook is idempotent: re-running for the same key produces the
* same row. When the day window holds zero events the row is deleted
* so the read path returns the trailing-window zero-default rather
* than a stale stub.
*
*/
export async function recomputeMedicationComplianceForDay(
userId: string,
medicationId: string,
dayKey: string,
tz: string | null | undefined,
client: MedicationComplianceRollupClient = prisma,
): Promise<void> {
const safeTz = safeTimezone(tz);
// The local day's own bounds: 23, 24 or 25 hours on DST days.
const { dayStart: start, dayEnd: windowEnd } = localDayWindow(dayKey, safeTz);
// v1.4.39: atomic upsert closes the race window
// between the prior `SELECT … aggregate then UPSERT` pattern. Two
// concurrent writes for the same (user, medication, day) used to
// interleave A-SELECT → B-SELECT → B-UPSERT (correct) → A-UPSERT
// (stale N-1). The INSERT … SELECT re-aggregates inside the upsert
// statement so each write commits with a snapshot taken after the
// prior commit released its row lock; the trailing DELETE handles
// the "all events for the day were removed" case in the same shot.
//
// Both statements key on (user_id, medication_id, day) so they
// serialise behind the row lock the rollup table's primary key
// provides. The DELETE runs second so it cannot clobber a row the
// INSERT just wrote — when the day still has events the
// `WHERE NOT EXISTS` predicate matches zero rows.
// v1.15.19 — slot-level aggregation. The inner CTE folds the rows that
// share one `scheduled_for` instant (cross-source duplicates: pending
// REMINDER + taken API) into one slot with taken-beats-skipped priority;
// the outer aggregate then counts slots, not rows, so a duplicated slot
// can never inflate `scheduled`. Pure SQL — re-running the recompute
// over historic duplicates self-corrects the counts.
await client.$executeRaw`
WITH slot AS (
SELECT
BOOL_OR("taken_at" IS NOT NULL AND NOT "skipped") AS slot_taken,
BOOL_OR("skipped") AS slot_skipped
FROM "medication_intake_events"
WHERE "user_id" = ${userId}
AND "medication_id" = ${medicationId}
AND "deleted_at" IS NULL
AND "scheduled_for" >= ${start}
AND "scheduled_for" < ${windowEnd}
GROUP BY "scheduled_for"
),
aggregate AS (
SELECT
COUNT(*)::int AS scheduled,
COALESCE(SUM(CASE WHEN "slot_taken" THEN 1 ELSE 0 END), 0)::int AS taken,
COALESCE(SUM(CASE WHEN NOT "slot_taken" AND "slot_skipped" THEN 1 ELSE 0 END), 0)::int AS skipped
FROM slot
)
INSERT INTO "medication_compliance_rollups"
("user_id", "medication_id", "day", "scheduled", "taken", "skipped", "computed_at")
SELECT
${userId},
${medicationId},
${dayKey},
aggregate.scheduled,
aggregate.taken,
aggregate.skipped,
NOW()
FROM aggregate
WHERE aggregate.scheduled > 0
ON CONFLICT ("user_id", "medication_id", "day") DO UPDATE
SET "scheduled" = EXCLUDED."scheduled",
"taken" = EXCLUDED."taken",
"skipped" = EXCLUDED."skipped",
"computed_at" = EXCLUDED."computed_at"
`;
await client.$executeRaw`
DELETE FROM "medication_compliance_rollups"
WHERE "user_id" = ${userId}
AND "medication_id" = ${medicationId}
AND "day" = ${dayKey}
AND NOT EXISTS (
SELECT 1
FROM "medication_intake_events"
WHERE "user_id" = ${userId}
AND "medication_id" = ${medicationId}
AND "deleted_at" IS NULL
AND "scheduled_for" >= ${start}
AND "scheduled_for" < ${windowEnd}
)
`;
// v1.43 — the day's other derived row: the micronutrients its taken
// supplement doses add (`source = SUPPLEMENT`). Every dose write path
// already funnels through this recompute, so this is the one place the
// hook needs to live. Only on the base client: a caller inside its own
// transaction (the dose import) has not committed the doses yet and
// queues the account's full recompute once it has.
if (client === prisma) {
await refreshSupplementNutrientsForDose({
userId,
medicationId,
dayKey,
timezone: safeTz,
});
}
}
/**
* Convenience hook for write-paths that have the `scheduledFor`
* instant in hand. Derives `dayKey` in the user's tz then dispatches
* to `recomputeMedicationComplianceForDay`. Best-effort: surfaces
* failures through `annotate` + `console.error` so a silent populator
* regression shows up in ops without blocking the parent write.
*/
export async function recomputeMedicationComplianceForEvent(input: {
userId: string;
medicationId: string;
scheduledFor: Date;
tz: string | null | undefined;
}): Promise<void> {
const dayKey = dayKeyForScheduledFor(input.scheduledFor, input.tz);
try {
await recomputeMedicationComplianceForDay(
input.userId,
input.medicationId,
dayKey,
input.tz,
);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
annotate({
meta: {
medication_compliance_rollup_failed: true,
medication_compliance_rollup_error: message,
},
});
console.error("[medication-compliance-rollups] recompute failed:", message);
}
}
/**
* Full-fold helper. Walks every `(medicationId, day)` pair the user
* has logged across the trailing `days` window and upserts each
* rollup row. The boot-time backfill worker calls this once per
* uncovered user; the route's lazy fallback enqueues it when the
* coverage probe returns false.
*
* Bounded SQL: one `DISTINCT (medicationId, dayKey)` $queryRaw scan
* over the trailing window, then one upsert per row. The dayKey is
* computed in Postgres in the user's zone so we
* don't have to ship millions of rows back to Node.
*/
export async function recomputeUserMedicationCompliance(
userId: string,
days: number = MEDICATION_COMPLIANCE_BACKFILL_DAYS,
tz: string | null | undefined = null,
): Promise<{ rowsUpserted: number; durationMs: number }> {
const startedAt = Date.now();
const safeTz = safeTimezone(tz);
const now = new Date();
const oldestKey = userDayKey(
new Date(now.getTime() - (days - 1) * 86_400_000),
safeTz,
);
const oldestStart = startOfLocalDayKey(oldestKey, safeTz);
// Distinct (medication_id, dayKey) pairs the user has any intake
// event for inside the trailing window. `scheduled_for` is a naive UTC
// timestamp: it is read as UTC first and only then shifted into the
// user's zone. A single `AT TIME ZONE $tz` reads it as local time and
// shifts it the wrong way, which keyed every slot within the zone's
// offset of midnight onto the neighbouring day.
const pairs = await prisma.$queryRaw<
Array<{ medication_id: string; day: string }>
>`
SELECT DISTINCT
"medication_id",
to_char(
("scheduled_for" AT TIME ZONE 'UTC') AT TIME ZONE ${safeTz},
'YYYY-MM-DD'
) AS "day"
FROM "medication_intake_events"
WHERE "user_id" = ${userId}
AND "deleted_at" IS NULL
AND "scheduled_for" >= ${oldestStart}
`;
let rowsUpserted = 0;
for (const pair of pairs) {
await recomputeMedicationComplianceForDay(
userId,
pair.medication_id,
pair.day,
safeTz,
);
rowsUpserted += 1;
}
return { rowsUpserted, durationMs: Date.now() - startedAt };
}
/**
* v1.4.39 — user-scoped enqueue helper for the
* request-path coverage-miss fallback.
*
* Mirrors `ensureUserMoodRollupsFresh`: the route fires this when its
* coverage probe returned false so the caller's account picks up a
* targeted boot-backfill job, instead of running a cluster-wide
* `LEFT JOIN` over `medication_intake_events × medication_compliance_rollups`
* on every authenticated coverage-miss request. The cluster-wide scan
* was a soft DoS amplifier — an authenticated user iterating
* coverage-miss requests could drive a multi-tenant table scan on
* every hit.
*
* The boot-time discovery helper below stays cluster-wide (it runs
* once per worker boot, not per request).
*/
export async function enqueueUserMedicationComplianceBackfill(
userId: string,
): Promise<{ enqueued: boolean; error: string | null }> {
const boss = getGlobalBoss();
if (!boss) {
return { enqueued: false, error: null };
}
try {
const payload: MedicationComplianceBackfillPayload = {
userId,
enqueuedAt: new Date().toISOString(),
};
const jobId = await boss.send(
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
payload,
{
retryLimit: 3,
retryDelay: 60,
retryBackoff: true,
singletonKey: `medication-compliance-boot-backfill|${userId}`,
},
);
return { enqueued: jobId !== null && jobId !== undefined, error: null };
} catch (err) {
return {
enqueued: false,
error: err instanceof Error ? err.message : String(err),
};
}
}
/**
* Boot-time enqueue helper — mirrors v1.4.35.1
* `enqueueBootTimeRollupBackfill`. Finds users with intake events but
* zero compliance-rollup coverage for the trailing window and enqueues
* one full-fold per account onto `MEDICATION_COMPLIANCE_BACKFILL_QUEUE`.
*
* Idempotent across reboots: once an account has at least one rollup
* row inside the trailing window the discovery query drops them.
* pg-boss `singletonKey` coalesces duplicate sends.
*/
export async function enqueueBootTimeMedicationComplianceBackfill(
// Optional boot-storm stagger. When > 0 the per-user sends carry a
// `startAfter` delay (seconds) so this full-history fold does not drain in
// parallel with the other boot backfills onto one heavy tenant. Default 0
// keeps immediate semantics for any non-boot caller.
startAfterSeconds: number = 0,
): Promise<{
enqueued: number;
skipped: number;
error: string | null;
}> {
const boss = getGlobalBoss();
if (!boss) {
return { enqueued: 0, skipped: 0, error: null };
}
try {
// Trailing window cutoff for the existence probe — `scheduled_for`
// older than this is not what the read path consults.
const cutoff = new Date(
Date.now() - MEDICATION_COMPLIANCE_BACKFILL_DAYS * 86_400_000,
);
const users = await prisma.$queryRaw<Array<{ id: string }>>`
SELECT DISTINCT mie."user_id" AS id
FROM "medication_intake_events" mie
LEFT JOIN "medication_compliance_rollups" r
ON r."user_id" = mie."user_id"
AND r."medication_id" = mie."medication_id"
WHERE mie."scheduled_for" >= ${cutoff}
AND mie."deleted_at" IS NULL
AND r."day" IS NULL
`;
if (users.length === 0) {
return { enqueued: 0, skipped: 0, error: null };
}
let enqueued = 0;
let skipped = 0;
for (const { id } of users) {
const payload: MedicationComplianceBackfillPayload = {
userId: id,
enqueuedAt: new Date().toISOString(),
};
const jobId = await boss.send(
MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
payload,
{
retryLimit: 3,
retryDelay: 60,
retryBackoff: true,
singletonKey: `medication-compliance-boot-backfill|${id}`,
...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}),
},
);
if (jobId) {
enqueued += 1;
} else {
skipped += 1;
}
}
return { enqueued, skipped, error: null };
} catch (err) {
return {
enqueued: 0,
skipped: 0,
error: err instanceof Error ? err.message : String(err),
};
}
}
+5 -5
View File
@@ -9,7 +9,7 @@ import {
import { getServerTranslator } from "@/lib/i18n/server-translator";
import { locales, type Locale } from "@/lib/i18n/config";
import { foldForMatch } from "@/lib/i18n/fold-for-match";
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh";
import {
applyCanonicalSlotWrite,
resolveForcedSlotForWrite,
@@ -212,8 +212,8 @@ async function findTelegramUser(chatId: string) {
id: true,
telegramBotToken: true,
locale: true,
// v1.4.39 — pulled into the select so the post-write
// compliance-rollup hook can anchor the user-day bucket.
// Pulled into the select so the post-write dose-day refresh can
// anchor the user's local day.
timezone: true,
},
});
@@ -428,7 +428,7 @@ async function markMedicationTaken(
// affected day so the next read after the cache miss reflects the
// new dose. Only fires when the write landed a row.
if (scheduledFor) {
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId,
medicationId: medication.id,
scheduledFor,
@@ -690,7 +690,7 @@ export async function handleCallback(update: TelegramUpdate) {
// event's user-day so the next read sees the (scheduled, skipped)
// counts incremented.
if (skippedScheduledFor) {
await recomputeMedicationComplianceForEvent({
await refreshDoseSlot({
userId: user.id,
medicationId: medication.id,
scheduledFor: skippedScheduledFor,
+2 -2
View File
@@ -1,6 +1,6 @@
/**
* Wall-clock-in-timezone helper shared by the medication scheduling
* + compliance-rollup paths.
* paths.
*
* Decomposes a `Date` instant into the year/month/day/hour/minute/
* second/weekday tuple an observer in `tz` would read off a wall
@@ -12,7 +12,7 @@
* per-user timezone work keep their original shape.
*
* v1.4.40 W-GHOSTS consolidated two file-local copies
* (`compliance-rollups.ts` returning `{year,month,day}` only +
* (a compliance-rollup copy returning `{year,month,day}` only +
* `scheduling/cadence.ts` returning the full tuple) into this single
* canonical export. The cross-tz ±3 h guard hinges on these helpers
* staying in lock-step; keeping them as separate definitions made
@@ -164,16 +164,6 @@ describe("intake-slot-dedup — duplicate slot collapse", () => {
});
expect(eveningRow.deletedAt).toBeNull();
expect(eveningRow.scheduledFor.toISOString()).toBe(slot1900.toISOString());
// The compliance rollup reflects the corrected counts: the 07:00 day
// now has scheduled=2 (07:00 taken + 19:00 pending), taken=1 — not the
// inflated scheduled=3 the duplicate produced.
const rollups = await prisma.medicationComplianceRollup.findMany({
where: { userId: TEST_USER_ID, medicationId: medId },
});
expect(rollups).toHaveLength(1);
expect(rollups[0]?.scheduled).toBe(2);
expect(rollups[0]?.taken).toBe(1);
});
it("never collapses a PRN med's multiple rows", async () => {
-29
View File
@@ -12,7 +12,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
import { getPrismaClient, truncateAllTables } from "./setup";
import { readActivityDayKeys } from "@/lib/dashboard/summary-reads";
import { recomputeUserMedicationCompliance } from "@/lib/rollups/medication-compliance-rollups";
import { userDayKey } from "@/lib/tz/format";
vi.mock("@/lib/db-compat", () => ({
@@ -94,31 +93,3 @@ describe("readActivityDayKeys", () => {
);
});
});
describe("recomputeUserMedicationCompliance", () => {
it("keys a late dose onto the person's own day", async () => {
const med = await prisma.medication.create({
data: { userId: USER, name: "Ramipril", dose: "5mg", active: true },
});
await prisma.medicationIntakeEvent.create({
data: {
userId: USER,
medicationId: med.id,
scheduledFor: LATE,
takenAt: LATE,
skipped: false,
source: "API",
},
});
await recomputeUserMedicationCompliance(USER, 30, "Europe/Berlin");
const rows = await prisma.medicationComplianceRollup.findMany({
where: { userId: USER },
select: { day: true, scheduled: true, taken: true },
});
expect(rows).toEqual([
{ day: key(D + 86_400_000), scheduled: 1, taken: 1 },
]);
});
});
@@ -76,13 +76,13 @@ vi.mock("@/lib/db-compat", () => ({
vi.mock("@/lib/cache/invalidate", () => ({
invalidateUserMedications: vi.fn(),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", async () => {
vi.mock("@/lib/medications/dose-day-refresh", async () => {
const actual = await vi.importActual<
typeof import("@/lib/rollups/medication-compliance-rollups")
>("@/lib/rollups/medication-compliance-rollups");
typeof import("@/lib/medications/dose-day-refresh")
>("@/lib/medications/dose-day-refresh");
return {
...actual,
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
refreshDoseSlot: vi.fn().mockResolvedValue(undefined),
};
});
@@ -1,225 +0,0 @@
/**
* v1.15.19 — slot-level medication-compliance rollup aggregation.
*
* The partial unique index on `(user, medication, scheduled_for, source)`
* lets two LIVE intake rows share one slot instant when their `source`
* differs — the production shape: the reminder worker pre-mints a pending
* REMINDER row on the slot, and a client write that the resolver refused
* to snap (future-slot guard) used to land a standalone API row on the
* exact same instant. The rollup previously counted rows (`COUNT(*)`), so
* a two-dose day with one duplicated slot reported `scheduled=3..4`.
*
* These tests pin the corrected semantics on real Postgres:
* 1. Two live rows on one slot, different source, one taken →
* scheduled=1, taken=1.
* 2. Taken beats skipped inside a slot.
* 3. The recompute self-corrects historic duplicates with no data
* migration (pure SQL, retroactive).
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
import { getPrismaClient, truncateAllTables } from "./setup";
import {
dayKeyForScheduledFor,
recomputeMedicationComplianceForDay,
} from "@/lib/rollups/medication-compliance-rollups";
const TEST_USER_ID = "user-compliance-rollup-slot";
const TZ = "Europe/Berlin";
vi.mock("@/lib/db-compat", () => ({
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
}));
let medId = "";
beforeEach(async () => {
const prisma = getPrismaClient();
await truncateAllTables(prisma);
await prisma.user.create({
data: {
id: TEST_USER_ID,
username: "compliance-rollup-slot",
email: "compliance-rollup-slot@example.test",
timezone: TZ,
},
});
const med = await prisma.medication.create({
data: {
userId: TEST_USER_ID,
name: "Ramipril",
dose: "5mg",
active: true,
},
});
medId = med.id;
});
// Fixed non-DST-boundary day so the slot instants are deterministic.
// 2026-03-10 07:00 / 19:00 Europe/Berlin (CET, UTC+1).
const SLOT_0700 = new Date("2026-03-10T06:00:00.000Z");
const SLOT_1900 = new Date("2026-03-10T18:00:00.000Z");
async function readRollup() {
const prisma = getPrismaClient();
const dayKey = dayKeyForScheduledFor(SLOT_0700, TZ);
await recomputeMedicationComplianceForDay(TEST_USER_ID, medId, dayKey, TZ);
return prisma.medicationComplianceRollup.findFirst({
where: { userId: TEST_USER_ID, medicationId: medId, day: dayKey },
});
}
describe("medication-compliance rollup — slot-level aggregation (real Postgres)", () => {
it("counts a cross-source duplicate slot once: scheduled=1, taken=1", async () => {
const prisma = getPrismaClient();
// The pending REMINDER row the worker minted on the slot…
await prisma.medicationIntakeEvent.create({
data: {
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: null,
skipped: false,
source: "REMINDER",
},
});
// …and the standalone API row a pre-fix client write parked on the
// exact same instant (the unique index tolerates it: source differs).
await prisma.medicationIntakeEvent.create({
data: {
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: new Date("2026-03-10T06:02:00.000Z"),
skipped: false,
source: "API",
},
});
const rollup = await readRollup();
expect(rollup).not.toBeNull();
expect(rollup?.scheduled).toBe(1);
expect(rollup?.taken).toBe(1);
expect(rollup?.skipped).toBe(0);
});
it("keeps a two-dose day at scheduled=2 when one slot is duplicated", async () => {
const prisma = getPrismaClient();
await prisma.medicationIntakeEvent.createMany({
data: [
// Duplicated morning slot: pending REMINDER + taken API.
{
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: null,
skipped: false,
source: "REMINDER",
},
{
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: new Date("2026-03-10T06:01:00.000Z"),
skipped: false,
source: "API",
},
// Clean evening slot, still pending.
{
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_1900,
takenAt: null,
skipped: false,
source: "REMINDER",
},
],
});
const rollup = await readRollup();
// Pre-fix this day reported scheduled=3 (row count). Slot-level it is 2.
expect(rollup?.scheduled).toBe(2);
expect(rollup?.taken).toBe(1);
expect(rollup?.skipped).toBe(0);
});
it("taken beats skipped inside one slot", async () => {
const prisma = getPrismaClient();
await prisma.medicationIntakeEvent.createMany({
data: [
{
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: null,
skipped: true,
source: "REMINDER",
},
{
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: new Date("2026-03-10T06:05:00.000Z"),
skipped: false,
source: "API",
},
],
});
const rollup = await readRollup();
expect(rollup?.scheduled).toBe(1);
expect(rollup?.taken).toBe(1);
expect(rollup?.skipped).toBe(0);
});
it("counts a purely skipped slot as skipped, not taken", async () => {
const prisma = getPrismaClient();
await prisma.medicationIntakeEvent.create({
data: {
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: null,
skipped: true,
source: "WEB",
},
});
const rollup = await readRollup();
expect(rollup?.scheduled).toBe(1);
expect(rollup?.taken).toBe(0);
expect(rollup?.skipped).toBe(1);
});
it("ignores soft-deleted rows when folding a slot", async () => {
const prisma = getPrismaClient();
await prisma.medicationIntakeEvent.createMany({
data: [
{
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: new Date("2026-03-10T06:02:00.000Z"),
skipped: false,
source: "API",
deletedAt: new Date(),
},
{
userId: TEST_USER_ID,
medicationId: medId,
scheduledFor: SLOT_0700,
takenAt: null,
skipped: false,
source: "REMINDER",
},
],
});
const rollup = await readRollup();
// The tombstoned taken row must not paint the live pending slot taken.
expect(rollup?.scheduled).toBe(1);
expect(rollup?.taken).toBe(0);
expect(rollup?.skipped).toBe(0);
});
});
@@ -286,24 +286,4 @@ describe("dose-history import — the reported export end to end", () => {
});
expect(stillInactive).toBe(16);
}, 240_000);
it("re-folds the compliance rollup for every medication and day it touched", async () => {
const prisma = getPrismaClient();
const user = await seedRegimen();
const { jobId } = await enqueue(user.id);
await processMedicationIntakeImportJob(jobId);
const rollups = await prisma.medicationComplianceRollup.findMany({
where: { userId: user.id },
select: { medicationId: true, day: true },
});
// One rollup row per (medication, local day) the import wrote into. The job
// spans sixteen medications, so a bare day key could not have addressed
// them — two medications on one day are two rows.
expect(rollups.length).toBeGreaterThan(3000);
expect(new Set(rollups.map((row) => row.medicationId)).size).toBe(16);
expect(
new Set(rollups.map((row) => `${row.medicationId} ${row.day}`)).size,
).toBe(rollups.length);
}, 480_000);
});
@@ -106,7 +106,7 @@ describe("medication intake import job — concurrent retry", () => {
]);
await processMedicationIntakeImportJob(job.id);
const [events, refreshedInventory, rollups, refreshedJob, audits] =
const [events, refreshedInventory, refreshedJob, audits] =
await Promise.all([
prisma.medicationIntakeEvent.findMany({
where: { userId: user.id, medicationId: medication.id },
@@ -115,10 +115,6 @@ describe("medication intake import job — concurrent retry", () => {
prisma.medicationInventoryItem.findUniqueOrThrow({
where: { id: inventory.id },
}),
prisma.medicationComplianceRollup.findMany({
where: { userId: user.id, medicationId: medication.id },
orderBy: { day: "asc" },
}),
prisma.medicationIntakeImportJob.findUniqueOrThrow({
where: { id: job.id },
}),
@@ -144,16 +140,6 @@ describe("medication intake import job — concurrent retry", () => {
),
).toBe(true);
expect(Number(refreshedInventory.unitsRemaining)).toBe(7);
expect(
rollups.map(({ day, scheduled, taken }) => ({
day,
scheduled,
taken,
})),
).toEqual([
{ day: "2026-07-20", scheduled: 2, taken: 2 },
{ day: "2026-07-21", scheduled: 1, taken: 1 },
]);
expect(refreshedJob.status).toBe("done");
// The two entries sharing an instant collapse inside the submission; that
// is `duplicate_in_file`, and it is reported under its own name rather
@@ -168,14 +154,10 @@ describe("medication intake import job — concurrent retry", () => {
total: 4,
imported: 3,
skippedByReason: { duplicate_in_file: 1 },
// A job can now cover a whole regimen, so a day whose rollup needs
// re-folding is addressed by medication AND day — two medications touched
// on one day are two rollup rows, which a bare day key cannot name.
touchedDays: [
`${medication.id} 2026-07-20`,
`${medication.id} 2026-07-21`,
],
rollupProcessed: 2,
// Inert since the compliance rollup was dropped: written at admission
// and never advanced.
touchedDays: [],
rollupProcessed: 0,
});
expect(audits).toHaveLength(1);
expect(invalidateUserMedications).toHaveBeenCalledTimes(1);
@@ -270,9 +252,6 @@ describe("medication intake import job — concurrent retry", () => {
prisma.medicationIntakeEvent.count({
where: { medicationId: medication.id },
}),
prisma.medicationComplianceRollup.count({
where: { medicationId: medication.id },
}),
prisma.auditLog.count({
where: {
userId: user.id,
@@ -285,20 +264,16 @@ describe("medication intake import job — concurrent retry", () => {
]);
expect(afterFailedFinalization[0]).toBe(1);
expect(afterFailedFinalization[1]).toBe(0);
expect(afterFailedFinalization[2]).toBe(0);
expect(afterFailedFinalization[3].status).toBe("running");
expect(afterFailedFinalization[2].status).toBe("running");
_setMedicationIntakeImportPrismaForTests(prisma);
await processMedicationIntakeImportJob(job.id);
await processMedicationIntakeImportJob(job.id);
const [events, rollups, audits, completedJob] = await Promise.all([
const [events, audits, completedJob] = await Promise.all([
prisma.medicationIntakeEvent.findMany({
where: { medicationId: medication.id },
}),
prisma.medicationComplianceRollup.findMany({
where: { medicationId: medication.id },
}),
prisma.auditLog.findMany({
where: {
userId: user.id,
@@ -310,7 +285,6 @@ describe("medication intake import job — concurrent retry", () => {
}),
]);
expect(events).toHaveLength(1);
expect(rollups).toHaveLength(1);
expect(audits).toHaveLength(1);
expect(completedJob.status).toBe("done");
});
@@ -14,9 +14,6 @@ vi.mock("@/lib/notifications/dispatcher", () => ({
channelsSucceeded: 1,
}),
}));
vi.mock("@/lib/rollups/medication-compliance-rollups", () => ({
recomputeMedicationComplianceForEvent: vi.fn().mockResolvedValue(undefined),
}));
const { dispatchNotification } = await import("@/lib/notifications/dispatcher");
const { handleReminderCheck } =
@@ -1337,30 +1337,4 @@ describe("marking a dose — what the owner is told, and what the rollup recompu
status: "taken",
});
});
it("recomputes the compliance rollup under the OWNER's id", async () => {
// The dose landed in the owner's record; a rollup row keyed on the
// delegate would leave the owner's compliance reading stale and give the
// delegate a row about a medication they do not have.
const owner = await makeUser("owner");
const delegate = await makeUser("delegate");
const { intakeId } = await seedPendingDose(owner.id);
await switchInto(owner.id, delegate.id, "WRITE");
const { POST } = await import("@/app/api/medications/intake/route");
expect(
(
await post(POST as Handler, "/api/medications/intake", {
intakeId,
status: "taken",
})
).status,
).toBe(200);
const rollups = await getPrismaClient().medicationComplianceRollup.findMany(
{ select: { userId: true } },
);
expect(rollups.length).toBeGreaterThan(0);
expect([...new Set(rollups.map((r) => r.userId))]).toEqual([owner.id]);
});
});
@@ -494,9 +494,9 @@ describe("taken doses become the day's SUPPLEMENT micronutrients", () => {
takenAt: twoDaysAgo,
},
});
const { recomputeMedicationComplianceForEvent } =
await import("@/lib/rollups/medication-compliance-rollups");
await recomputeMedicationComplianceForEvent({
const { refreshDoseSlot } =
await import("@/lib/medications/dose-day-refresh");
await refreshDoseSlot({
userId,
medicationId,
scheduledFor: twoDaysAgo,