mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 20:42:43 +00:00
fix(medications): collapse pre-existing duplicate dose-slot rows at boot
The write-path convergence stops new duplicate slot rows, but accounts that already accrued a pending + taken pair for one dose keep an inflated expected count, a phantom taken marker, and a suppressed "due now" prompt until the bad rows are cleared. Add a boot-time, idempotent cleanup queue (mirroring the existing backfill queues): discover only the users holding a same-slot duplicate, then per medication snap every live intake to its canonical slot, keep the most-actioned row per slot (taken over skipped over pending), normalise its instant, soft-delete the losers, and recompute the affected compliance rollups. Rows that resolve to no scheduled slot (PRN / as-needed) are never collapsed. A second pass is a no-op. Registered in the worker's queue list with a static guard test, after a prior incident where a queue was built but never registered and silently never ran.
This commit is contained in:
1 parent
0d7d08f616
commit
5047eb7a20
4 files changed
+877
No files matched your search
@@ -0,0 +1,46 @@
|
||||
/**
|
||||
* v1.8.2 — duplicate dose-slot dedup queue registration guard.
|
||||
*
|
||||
* Same source-text-grep approach as the drain + step-consolidation +
|
||||
* mean-consolidation regression guards: assert the queue is registered
|
||||
* in `allQueues`, a `boss.work` handler is wired against it, and the
|
||||
* boot-discovery enqueue helper is called — without booting pg-boss +
|
||||
* Prisma. An unregistered queue silently never drains (the past incident
|
||||
* this guard exists to prevent).
|
||||
*/
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
|
||||
const source = readFileSync(REMINDER_WORKER_PATH, "utf8");
|
||||
|
||||
describe("reminder-worker — intake-slot-dedup wiring", () => {
|
||||
it("imports the queue symbols from the intake-slot-dedup module", () => {
|
||||
expect(source).toMatch(
|
||||
/from\s*["']@\/lib\/medications\/intake-slot-dedup["']/,
|
||||
);
|
||||
expect(source).toMatch(/\bINTAKE_SLOT_DEDUP_QUEUE\b/);
|
||||
expect(source).toMatch(/\bdedupeUserIntakeSlots\b/);
|
||||
expect(source).toMatch(/\benqueueBootTimeIntakeSlotDedup\b/);
|
||||
});
|
||||
|
||||
it("registers the intake-slot-dedup queue in the allQueues loop", () => {
|
||||
const allQueuesMatch = source.match(
|
||||
/const allQueues\s*=\s*\[([\s\S]*?)\];/,
|
||||
);
|
||||
expect(allQueuesMatch).not.toBeNull();
|
||||
expect(allQueuesMatch![1]).toMatch(/\bINTAKE_SLOT_DEDUP_QUEUE\b/);
|
||||
});
|
||||
|
||||
it("registers a boss.work handler against the intake-slot-dedup queue", () => {
|
||||
expect(source).toMatch(
|
||||
/boss\.work[\s\S]{0,200}INTAKE_SLOT_DEDUP_QUEUE[\s\S]{0,400}dedupeUserIntakeSlots/,
|
||||
);
|
||||
});
|
||||
|
||||
it("fires the boot-discovery enqueue helper", () => {
|
||||
expect(source).toMatch(/await enqueueBootTimeIntakeSlotDedup\(\)/);
|
||||
});
|
||||
});
|
||||
@@ -126,6 +126,13 @@ import {
|
||||
enqueueBootTimeMeanConsolidation,
|
||||
type MeanConsolidationPayload,
|
||||
} from "@/lib/jobs/mean-consolidation";
|
||||
import {
|
||||
INTAKE_SLOT_DEDUP_QUEUE,
|
||||
INTAKE_SLOT_DEDUP_CONCURRENCY,
|
||||
dedupeUserIntakeSlots,
|
||||
enqueueBootTimeIntakeSlotDedup,
|
||||
type IntakeSlotDedupPayload,
|
||||
} from "@/lib/medications/intake-slot-dedup";
|
||||
import { expireStaleInUseItems } from "@/lib/medications/inventory/service";
|
||||
import { rotateLegacyMoodLogSecrets } from "@/lib/moodlog-secret";
|
||||
import { probeIntegrationStatusNullBuckets } from "@/lib/jobs/integration-status-null-probe";
|
||||
@@ -2017,6 +2024,16 @@ export async function startReminderWorker() {
|
||||
// reboots. The queue MUST be registered here or pg-boss never
|
||||
// provisions it and the boot enqueue silently never drains.
|
||||
MEAN_CONSOLIDATION_QUEUE,
|
||||
// v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery
|
||||
// enqueues one job per user holding two live intake rows that snap to
|
||||
// the same canonical slot (the pre-fix REMINDER-pending + API-taken
|
||||
// pair). The per-user pass keeps the winner (taken > skipped >
|
||||
// pending), soft-deletes the losers, and recomputes the affected
|
||||
// compliance rollups. Idempotent across reboots — a deduped user
|
||||
// falls off the discovery list. The queue MUST be registered here or
|
||||
// pg-boss never provisions it and the boot enqueue silently never
|
||||
// drains.
|
||||
INTAKE_SLOT_DEDUP_QUEUE,
|
||||
// v1.4.37 W7c — explicit createQueue is required before the
|
||||
// nightly schedule below registers (pg-boss v12 contract). Without
|
||||
// this entry the drain schedule silently no-ops and the
|
||||
@@ -2397,6 +2414,37 @@ export async function startReminderWorker() {
|
||||
},
|
||||
);
|
||||
|
||||
// v1.8.2 — duplicate dose-slot cleanup worker. The boot enqueue helper
|
||||
// below sends one job per user holding two live intake rows that snap
|
||||
// to the same canonical slot; this handler keeps the winner, soft-
|
||||
// deletes the losers, normalises the winner's scheduledFor, and
|
||||
// recomputes the affected compliance rollups. Serial concurrency so the
|
||||
// one-time pass never crowds the request pool.
|
||||
await boss.work<IntakeSlotDedupPayload>(
|
||||
INTAKE_SLOT_DEDUP_QUEUE,
|
||||
{ localConcurrency: INTAKE_SLOT_DEDUP_CONCURRENCY },
|
||||
async (jobs) => {
|
||||
for (const job of jobs) {
|
||||
const { userId } = job.data;
|
||||
try {
|
||||
const summary = await dedupeUserIntakeSlots(userId);
|
||||
workerLog(
|
||||
"info",
|
||||
`[intake-slot-dedup] user=${userId} slotsCollapsed=${summary.slotsCollapsed} rowsSoftDeleted=${summary.rowsSoftDeleted} rowsNormalised=${summary.rowsNormalised} daysRecomputed=${summary.daysRecomputed}`,
|
||||
);
|
||||
} catch (err) {
|
||||
recordError();
|
||||
workerLog(
|
||||
"error",
|
||||
`[intake-slot-dedup] user=${userId} failed`,
|
||||
err,
|
||||
);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
// v1.4.39 W-MOOD — mood-rollup per-bucket worker. Folds the
|
||||
// WEEK / MONTH / YEAR buckets that the mood-entry write hooks
|
||||
// enqueue; the DAY bucket runs synchronously in the hook itself.
|
||||
@@ -2602,6 +2650,35 @@ export async function startReminderWorker() {
|
||||
);
|
||||
}
|
||||
|
||||
// v1.8.2 — fire-and-forget boot discovery for the duplicate dose-slot
|
||||
// cleanup. Finds every user holding two live intake rows within the
|
||||
// drift window on the same medication and enqueues one dedup job per
|
||||
// account. Idempotent across reboots: collapsed losers are soft-deleted
|
||||
// so the `deleted_at IS NULL` discovery predicate drops them. Errors are
|
||||
// returned through the helper's result value — the worker boot never
|
||||
// fails because of a dedup miss.
|
||||
try {
|
||||
const { enqueued, skipped, error } =
|
||||
await enqueueBootTimeIntakeSlotDedup();
|
||||
if (error) {
|
||||
workerLog(
|
||||
"error",
|
||||
`[intake-slot-dedup] boot discovery failed: ${error}`,
|
||||
);
|
||||
} else {
|
||||
workerLog(
|
||||
"info",
|
||||
`[intake-slot-dedup] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
workerLog(
|
||||
"error",
|
||||
"[intake-slot-dedup] boot discovery threw an unexpected error",
|
||||
err,
|
||||
);
|
||||
}
|
||||
|
||||
// v1.4.39 W-MOOD — fire-and-forget boot discovery for the mood
|
||||
// rollup tier. Mirrors the v1.4.35.1 measurement-rollup pattern:
|
||||
// one job per user with mood entries but no rollup coverage.
|
||||
|
||||
@@ -0,0 +1,402 @@
|
||||
/**
|
||||
* v1.8.2 — one-time, idempotent cleanup for duplicate dose-slot rows.
|
||||
*
|
||||
* The v1.8.2 write-path fix (`resolveCanonicalSlotInstant` +
|
||||
* `shouldMintMissedDoseRow`) converges every NEW intake write onto one
|
||||
* canonical row per scheduled slot. Existing user data, written before
|
||||
* that fix, still carries duplicates: a twice-daily med ends up with a
|
||||
* 07:00 pending REMINDER row AND a 07:00 taken WEB/API row for the same
|
||||
* dose slot, with the two `scheduledFor` instants drifting by up to a
|
||||
* minute. The duplicate inflates the per-day `scheduled` count, paints a
|
||||
* phantom "taken", and confuses the "due now" prompt.
|
||||
*
|
||||
* This module is the boot-time backfill that collapses those existing
|
||||
* duplicates. It mirrors the `rollup-full-backfill` / `step-consolidation`
|
||||
* boot pattern:
|
||||
* - a cheap discovery pre-query enqueues one job ONLY for users that
|
||||
* actually hold duplicate slot rows;
|
||||
* - the per-user handler groups live `MedicationIntakeEvent` rows into
|
||||
* dose slots (reusing `resolveCanonicalSlotInstant` for the snap),
|
||||
* picks one winner per slot (taken > skipped > pending), normalises
|
||||
* the winner's `scheduledFor` to the canonical slot instant, and
|
||||
* soft-deletes the losers (NEVER hard-deletes — preserves the sync
|
||||
* tombstone contract for iOS delta-sync);
|
||||
* - it then recomputes the affected `(medicationId, day)` compliance
|
||||
* rollups so the scheduled/taken counts self-correct;
|
||||
* - it is idempotent across reboots: once a user's duplicates are
|
||||
* collapsed, the discovery pre-query no longer matches them and a
|
||||
* re-run of the per-user pass is a no-op.
|
||||
*
|
||||
* PRN / off-slot / as-needed rows resolve to `null` under
|
||||
* `resolveCanonicalSlotInstant` and are NEVER collapsed — they keep their
|
||||
* own rows untouched.
|
||||
*
|
||||
* Medical safety: when a slot has a TAKEN and a still-PENDING row we keep
|
||||
* the TAKEN. Deleting a recorded dose would falsely under-report
|
||||
* adherence — the dangerous direction. This matches the iOS team's
|
||||
* guidance.
|
||||
*
|
||||
* The queue name MUST be registered in `allQueues` in
|
||||
* `src/lib/jobs/reminder-worker.ts` so pg-boss provisions it at boot; an
|
||||
* unregistered queue silently never drains.
|
||||
*/
|
||||
import { prisma } from "@/lib/db";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { DEFAULT_TIMEZONE } from "@/lib/tz/format";
|
||||
import { resolveCanonicalSlotInstant } from "@/lib/medications/scheduling/resolve-slot-instant";
|
||||
import type {
|
||||
WorkerMedicationRow,
|
||||
WorkerScheduleRow,
|
||||
} from "@/lib/medications/scheduling/worker-helpers";
|
||||
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
|
||||
|
||||
export const INTAKE_SLOT_DEDUP_QUEUE = "intake-slot-dedup";
|
||||
|
||||
/**
|
||||
* Serial concurrency — the populator walks a user's intake events and
|
||||
* writes per slot; concurrency-1 keeps it from crowding the request
|
||||
* pool, matching the other boot-time backfills.
|
||||
*/
|
||||
export const INTAKE_SLOT_DEDUP_CONCURRENCY = 1;
|
||||
|
||||
export interface IntakeSlotDedupPayload {
|
||||
userId: string;
|
||||
enqueuedAt: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-user summary returned by `dedupeUserIntakeSlots` so the worker can
|
||||
* log the counts.
|
||||
*/
|
||||
export interface IntakeSlotDedupSummary {
|
||||
/** Distinct canonical slots that held more than one live row. */
|
||||
slotsCollapsed: number;
|
||||
/** Loser rows soft-deleted across all collapsed slots. */
|
||||
rowsSoftDeleted: number;
|
||||
/** Winner rows whose `scheduledFor` was normalised to the slot instant. */
|
||||
rowsNormalised: number;
|
||||
/** `(medicationId, dayKey)` pairs whose compliance rollup was recomputed. */
|
||||
daysRecomputed: number;
|
||||
}
|
||||
|
||||
/** The minimal medication+schedule projection the snap resolver needs. */
|
||||
type DedupMedication = WorkerMedicationRow & {
|
||||
userId: string;
|
||||
timezone: string;
|
||||
schedules: WorkerScheduleRow[];
|
||||
};
|
||||
|
||||
interface DedupIntakeRow {
|
||||
id: string;
|
||||
medicationId: string;
|
||||
scheduledFor: Date;
|
||||
takenAt: Date | null;
|
||||
skipped: boolean;
|
||||
syncVersion: number;
|
||||
createdAt: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* Winner priority of a row inside a slot: taken (2) > skipped (1) >
|
||||
* pending (0). A higher number wins.
|
||||
*/
|
||||
function rowPriority(row: DedupIntakeRow): number {
|
||||
if (row.takenAt !== null) return 2;
|
||||
if (row.skipped) return 1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pick the winner among the rows that share one canonical slot.
|
||||
* 1. Highest priority (taken > skipped > pending).
|
||||
* 2. Tie-break: most recent — `syncVersion` desc, then `createdAt`
|
||||
* desc, then `id` desc (lexicographic, deterministic).
|
||||
*/
|
||||
function pickWinner(rows: DedupIntakeRow[]): DedupIntakeRow {
|
||||
return rows.reduce((best, candidate) => {
|
||||
const bp = rowPriority(best);
|
||||
const cp = rowPriority(candidate);
|
||||
if (cp !== bp) return cp > bp ? candidate : best;
|
||||
if (candidate.syncVersion !== best.syncVersion) {
|
||||
return candidate.syncVersion > best.syncVersion ? candidate : best;
|
||||
}
|
||||
if (candidate.createdAt.getTime() !== best.createdAt.getTime()) {
|
||||
return candidate.createdAt.getTime() > best.createdAt.getTime()
|
||||
? candidate
|
||||
: best;
|
||||
}
|
||||
return candidate.id > best.id ? candidate : best;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Collapse duplicate dose-slot rows for one user. Best-effort and
|
||||
* idempotent: a second run finds zero multi-row slots and does nothing.
|
||||
*
|
||||
* For each of the user's medications, each live intake row is snapped to
|
||||
* its canonical scheduled-slot instant via `resolveCanonicalSlotInstant`.
|
||||
* Rows that snap to `null` (PRN / off-slot / as-needed) are left
|
||||
* untouched. Rows that snap to the same instant form a slot; a slot with
|
||||
* more than one row is collapsed: the winner's `scheduledFor` is
|
||||
* normalised to the canonical instant and the losers are soft-deleted.
|
||||
* Affected `(medicationId, dayKey)` compliance rollups are recomputed.
|
||||
*/
|
||||
export async function dedupeUserIntakeSlots(
|
||||
userId: string,
|
||||
): Promise<IntakeSlotDedupSummary> {
|
||||
const summary: IntakeSlotDedupSummary = {
|
||||
slotsCollapsed: 0,
|
||||
rowsSoftDeleted: 0,
|
||||
rowsNormalised: 0,
|
||||
daysRecomputed: 0,
|
||||
};
|
||||
|
||||
const user = await prisma.user.findUnique({
|
||||
where: { id: userId },
|
||||
select: { timezone: true },
|
||||
});
|
||||
const userTz = user?.timezone || DEFAULT_TIMEZONE;
|
||||
|
||||
const medications = (await prisma.medication.findMany({
|
||||
where: { userId },
|
||||
select: {
|
||||
id: true,
|
||||
userId: true,
|
||||
startsOn: true,
|
||||
endsOn: true,
|
||||
oneShot: true,
|
||||
createdAt: true,
|
||||
schedules: {
|
||||
select: {
|
||||
id: true,
|
||||
windowStart: true,
|
||||
windowEnd: true,
|
||||
daysOfWeek: true,
|
||||
timesOfDay: true,
|
||||
reminderGraceMinutes: true,
|
||||
rrule: true,
|
||||
rollingIntervalDays: true,
|
||||
scheduleType: true,
|
||||
cyclicOnWeeks: true,
|
||||
cyclicOffWeeks: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
})) as unknown as Array<DedupMedication & { timezone?: string }>;
|
||||
|
||||
// `(medicationId, dayKey)` pairs whose compliance rollup needs a
|
||||
// recompute after the collapse. Deduped so a busy day folds once.
|
||||
const daysToRecompute = new Set<string>();
|
||||
|
||||
for (const med of medications) {
|
||||
if (med.schedules.length === 0) continue;
|
||||
|
||||
// Latest non-tombstoned takenAt — only rolling schedules consult it,
|
||||
// but the snap resolver needs it threaded in.
|
||||
const lastIntake = await prisma.medicationIntakeEvent.findFirst({
|
||||
where: {
|
||||
userId,
|
||||
medicationId: med.id,
|
||||
deletedAt: null,
|
||||
takenAt: { not: null },
|
||||
},
|
||||
orderBy: { takenAt: "desc" },
|
||||
select: { takenAt: true },
|
||||
});
|
||||
const lastIntakeAt = lastIntake?.takenAt ?? null;
|
||||
|
||||
const rows = (await prisma.medicationIntakeEvent.findMany({
|
||||
where: { userId, medicationId: med.id, deletedAt: null },
|
||||
select: {
|
||||
id: true,
|
||||
medicationId: true,
|
||||
scheduledFor: true,
|
||||
takenAt: true,
|
||||
skipped: true,
|
||||
syncVersion: true,
|
||||
createdAt: true,
|
||||
},
|
||||
orderBy: { scheduledFor: "asc" },
|
||||
})) as DedupIntakeRow[];
|
||||
|
||||
if (rows.length < 2) continue;
|
||||
|
||||
// Group rows by the canonical slot instant they snap to. Rows that
|
||||
// resolve to `null` are PRN / off-slot — keyed individually so they
|
||||
// never collapse together.
|
||||
const slots = new Map<string, { instant: Date; rows: DedupIntakeRow[] }>();
|
||||
for (const row of rows) {
|
||||
const canonical = resolveCanonicalSlotInstant({
|
||||
medication: {
|
||||
id: med.id,
|
||||
startsOn: med.startsOn,
|
||||
endsOn: med.endsOn,
|
||||
oneShot: med.oneShot,
|
||||
createdAt: med.createdAt,
|
||||
schedules: med.schedules,
|
||||
},
|
||||
userTz,
|
||||
incoming: row.scheduledFor,
|
||||
lastIntakeAt,
|
||||
});
|
||||
if (canonical === null) continue; // PRN / off-slot — never collapse.
|
||||
const key = canonical.toISOString();
|
||||
const bucket = slots.get(key);
|
||||
if (bucket) {
|
||||
bucket.rows.push(row);
|
||||
} else {
|
||||
slots.set(key, { instant: canonical, rows: [row] });
|
||||
}
|
||||
}
|
||||
|
||||
for (const { instant, rows: slotRows } of slots.values()) {
|
||||
if (slotRows.length < 2) continue; // already one row — nothing to do.
|
||||
|
||||
const winner = pickWinner(slotRows);
|
||||
const losers = slotRows.filter((r) => r.id !== winner.id);
|
||||
|
||||
// Soft-delete every loser in one shot. Bump syncVersion so the
|
||||
// sync feed echoes a monotonic value and iOS drops the tombstone.
|
||||
const loserIds = losers.map((r) => r.id);
|
||||
if (loserIds.length > 0) {
|
||||
await prisma.medicationIntakeEvent.updateMany({
|
||||
where: { id: { in: loserIds }, deletedAt: null },
|
||||
data: { deletedAt: new Date(), syncVersion: { increment: 1 } },
|
||||
});
|
||||
summary.rowsSoftDeleted += loserIds.length;
|
||||
}
|
||||
|
||||
// Normalise the winner's scheduledFor to the canonical instant so
|
||||
// future writes upsert onto it. Skip the write (and the
|
||||
// syncVersion bump) when it already sits on the canonical instant.
|
||||
if (winner.scheduledFor.getTime() !== instant.getTime()) {
|
||||
await prisma.medicationIntakeEvent.update({
|
||||
where: { id: winner.id },
|
||||
data: { scheduledFor: instant, syncVersion: { increment: 1 } },
|
||||
});
|
||||
summary.rowsNormalised += 1;
|
||||
}
|
||||
|
||||
summary.slotsCollapsed += 1;
|
||||
|
||||
// The canonical instant determines the rollup day-key.
|
||||
daysToRecompute.add(`${med.id}|${instant.toISOString()}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Recompute the affected compliance rollups so scheduled/taken counts
|
||||
// and the rate self-correct. Best-effort per the helper's contract.
|
||||
for (const key of daysToRecompute) {
|
||||
const sep = key.indexOf("|");
|
||||
const medicationId = key.slice(0, sep);
|
||||
const instantIso = key.slice(sep + 1);
|
||||
await recomputeMedicationComplianceForEvent({
|
||||
userId,
|
||||
medicationId,
|
||||
scheduledFor: new Date(instantIso),
|
||||
tz: userTz,
|
||||
});
|
||||
summary.daysRecomputed += 1;
|
||||
}
|
||||
|
||||
annotate({
|
||||
action: {
|
||||
name: "medication.intake.slot_dedup",
|
||||
details: {
|
||||
user_id: userId,
|
||||
slots_collapsed: summary.slotsCollapsed,
|
||||
rows_soft_deleted: summary.rowsSoftDeleted,
|
||||
rows_normalised: summary.rowsNormalised,
|
||||
days_recomputed: summary.daysRecomputed,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
return summary;
|
||||
}
|
||||
|
||||
/**
|
||||
* Boot-time discovery. Finds every user that holds more than one live
|
||||
* `MedicationIntakeEvent` row that COULD be a duplicate slot — i.e. a
|
||||
* `(medicationId, scheduledFor-truncated-to-the-minute)` cluster with
|
||||
* more than one row, OR two live rows on the same medication within a
|
||||
* few minutes of each other (the iOS-vs-server drift). Enqueues one
|
||||
* dedup job per matching account.
|
||||
*
|
||||
* The pre-query is intentionally a cheap over-approximation: it matches
|
||||
* any user with two live rows on the same medication whose `scheduledFor`
|
||||
* fall in the same minute OR within a 2-minute window. The per-user pass
|
||||
* does the precise canonical-slot grouping and is a no-op when the rows
|
||||
* turn out not to share a slot, so a false-positive in discovery costs
|
||||
* one idle job, never a wrong collapse.
|
||||
*
|
||||
* Idempotent across reboots: once a user's duplicates are collapsed the
|
||||
* losers are soft-deleted (`deleted_at IS NOT NULL`) so the
|
||||
* `deleted_at IS NULL` predicate drops them and the user falls off the
|
||||
* list. pg-boss `singletonKey` coalesces duplicate sends.
|
||||
*
|
||||
* Best-effort: errors are returned through the result value so the
|
||||
* worker boot never fails because of a dedup miss.
|
||||
*/
|
||||
export async function enqueueBootTimeIntakeSlotDedup(): Promise<{
|
||||
enqueued: number;
|
||||
skipped: number;
|
||||
error: string | null;
|
||||
}> {
|
||||
const boss = getGlobalBoss();
|
||||
if (!boss) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
}
|
||||
|
||||
try {
|
||||
// Two live rows on the same medication whose scheduledFor land within
|
||||
// a 2-minute window are duplicate-slot candidates. `date_trunc`'d
|
||||
// self-join keeps the scan index-friendly on
|
||||
// `(user_id, medication_id, scheduled_for)`. The window absorbs the
|
||||
// sub-minute iOS-vs-server drift; the per-user pass re-checks
|
||||
// precisely so a wider-than-needed window only costs idle jobs.
|
||||
const users = await prisma.$queryRaw<Array<{ id: string }>>`
|
||||
SELECT DISTINCT a."user_id" AS id
|
||||
FROM "medication_intake_events" a
|
||||
JOIN "medication_intake_events" b
|
||||
ON a."user_id" = b."user_id"
|
||||
AND a."medication_id" = b."medication_id"
|
||||
AND a."id" <> b."id"
|
||||
AND a."deleted_at" IS NULL
|
||||
AND b."deleted_at" IS NULL
|
||||
AND abs(extract(epoch FROM (a."scheduled_for" - b."scheduled_for"))) <= 120
|
||||
`;
|
||||
|
||||
if (users.length === 0) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
}
|
||||
|
||||
let enqueued = 0;
|
||||
let skipped = 0;
|
||||
for (const { id } of users) {
|
||||
const payload: IntakeSlotDedupPayload = {
|
||||
userId: id,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
};
|
||||
const jobId = await boss.send(INTAKE_SLOT_DEDUP_QUEUE, payload, {
|
||||
retryLimit: 3,
|
||||
retryDelay: 60,
|
||||
retryBackoff: true,
|
||||
singletonKey: `intake-slot-dedup|${id}`,
|
||||
});
|
||||
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),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,352 @@
|
||||
/**
|
||||
* v1.8.2 — duplicate dose-slot cleanup integration tests.
|
||||
*
|
||||
* Pins the one-time boot-backfill that collapses pre-fix duplicate slot
|
||||
* rows: a twice-daily med ends up with a 07:00 pending REMINDER row AND
|
||||
* a 07:00 taken WEB/API row for the same dose slot (the two
|
||||
* `scheduledFor` instants drifting by up to a minute). The cleanup keeps
|
||||
* the winner (taken > skipped > pending), soft-deletes the losers,
|
||||
* normalises the winner's `scheduledFor` to the canonical slot instant,
|
||||
* leaves PRN / off-slot rows untouched, and is a no-op on a second run.
|
||||
*
|
||||
* Coverage map:
|
||||
* 1. Duplicate 07:00 pair collapses to the taken row; loser tombstoned;
|
||||
* the clean 19:00 row is untouched; compliance rollup self-corrects.
|
||||
* 2. A PRN med's multiple rows are never collapsed.
|
||||
* 3. A second run is a no-op (idempotent).
|
||||
* 4. taken > skipped tie-break keeps the taken row.
|
||||
* 5. Boot discovery only matches users that actually hold duplicates.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
import {
|
||||
dedupeUserIntakeSlots,
|
||||
enqueueBootTimeIntakeSlotDedup,
|
||||
} from "@/lib/medications/intake-slot-dedup";
|
||||
import { localHmAsUtc } from "@/lib/timezone";
|
||||
|
||||
const TEST_USER_ID = "user-intake-slot-dedup";
|
||||
const OTHER_USER_ID = "user-intake-slot-dedup-clean";
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
beforeEach(async () => {
|
||||
const prisma = getPrismaClient();
|
||||
await truncateAllTables(prisma);
|
||||
await prisma.user.create({
|
||||
data: {
|
||||
id: TEST_USER_ID,
|
||||
username: "intake-dedup",
|
||||
email: "intake-dedup@example.test",
|
||||
timezone: "Europe/Berlin",
|
||||
},
|
||||
});
|
||||
await prisma.user.create({
|
||||
data: {
|
||||
id: OTHER_USER_ID,
|
||||
username: "intake-dedup-clean",
|
||||
email: "intake-dedup-clean@example.test",
|
||||
timezone: "Europe/Berlin",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Anchor every fixture on a fixed historical day so the slot instants
|
||||
* are deterministic regardless of when the suite runs. 2026-03-10 is a
|
||||
* non-DST-boundary weekday in Europe/Berlin.
|
||||
*/
|
||||
const DAY = new Date("2026-03-10T12:00:00.000Z");
|
||||
|
||||
async function createTwiceDailyMed(): Promise<string> {
|
||||
const prisma = getPrismaClient();
|
||||
const med = await prisma.medication.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
name: "Bisoprolol",
|
||||
dose: "2.5mg",
|
||||
active: true,
|
||||
schedules: {
|
||||
create: {
|
||||
windowStart: "07:00",
|
||||
windowEnd: "07:00",
|
||||
timesOfDay: ["07:00", "19:00"],
|
||||
daysOfWeek: null,
|
||||
scheduleType: "SCHEDULED",
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
return med.id;
|
||||
}
|
||||
|
||||
describe("intake-slot-dedup — duplicate slot collapse", () => {
|
||||
it("collapses the 07:00 pending+taken pair to the taken row, leaves 19:00 untouched, corrects the rollup", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const medId = await createTwiceDailyMed();
|
||||
|
||||
const slot0700 = localHmAsUtc(DAY, "Europe/Berlin", 7, 0);
|
||||
const slot1900 = localHmAsUtc(DAY, "Europe/Berlin", 19, 0);
|
||||
|
||||
// 07:00 pending REMINDER row (server-minted, canonical instant).
|
||||
const pending = await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
scheduledFor: slot0700,
|
||||
takenAt: null,
|
||||
skipped: false,
|
||||
source: "REMINDER",
|
||||
},
|
||||
});
|
||||
// 07:00 taken API row, drifted +1 minute (the iOS-vs-server drift).
|
||||
const taken = await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
scheduledFor: new Date(slot0700.getTime() + 60_000),
|
||||
takenAt: new Date(slot0700.getTime() + 60_000),
|
||||
skipped: false,
|
||||
source: "API",
|
||||
},
|
||||
});
|
||||
// Clean 19:00 pending row — no duplicate.
|
||||
const evening = await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
scheduledFor: slot1900,
|
||||
takenAt: null,
|
||||
skipped: false,
|
||||
source: "REMINDER",
|
||||
},
|
||||
});
|
||||
|
||||
const summary = await dedupeUserIntakeSlots(TEST_USER_ID);
|
||||
|
||||
expect(summary.slotsCollapsed).toBe(1);
|
||||
expect(summary.rowsSoftDeleted).toBe(1);
|
||||
|
||||
// Exactly one live 07:00 row survives, and it is the taken one.
|
||||
const live0700 = await prisma.medicationIntakeEvent.findMany({
|
||||
where: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
deletedAt: null,
|
||||
scheduledFor: slot0700,
|
||||
},
|
||||
});
|
||||
expect(live0700).toHaveLength(1);
|
||||
expect(live0700[0]?.id).toBe(taken.id);
|
||||
expect(live0700[0]?.takenAt).not.toBeNull();
|
||||
// Winner normalised onto the canonical slot instant.
|
||||
expect(live0700[0]?.scheduledFor.toISOString()).toBe(
|
||||
slot0700.toISOString(),
|
||||
);
|
||||
|
||||
// The loser (the pending REMINDER row) is soft-deleted, not removed.
|
||||
const loser = await prisma.medicationIntakeEvent.findUniqueOrThrow({
|
||||
where: { id: pending.id },
|
||||
});
|
||||
expect(loser.deletedAt).not.toBeNull();
|
||||
|
||||
// The clean 19:00 row is untouched.
|
||||
const eveningRow = await prisma.medicationIntakeEvent.findUniqueOrThrow({
|
||||
where: { id: evening.id },
|
||||
});
|
||||
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 () => {
|
||||
const prisma = getPrismaClient();
|
||||
const med = await prisma.medication.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
name: "Ibuprofen PRN",
|
||||
dose: "400mg",
|
||||
active: true,
|
||||
schedules: {
|
||||
create: {
|
||||
windowStart: "00:00",
|
||||
windowEnd: "00:00",
|
||||
timesOfDay: [],
|
||||
daysOfWeek: null,
|
||||
scheduleType: "PRN",
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
// Two as-needed doses logged minutes apart — both must survive.
|
||||
const a = await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: med.id,
|
||||
scheduledFor: new Date("2026-03-10T09:00:00.000Z"),
|
||||
takenAt: new Date("2026-03-10T09:00:00.000Z"),
|
||||
skipped: false,
|
||||
source: "WEB",
|
||||
},
|
||||
});
|
||||
const b = await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: med.id,
|
||||
scheduledFor: new Date("2026-03-10T09:01:00.000Z"),
|
||||
takenAt: new Date("2026-03-10T09:01:00.000Z"),
|
||||
skipped: false,
|
||||
source: "WEB",
|
||||
},
|
||||
});
|
||||
|
||||
const summary = await dedupeUserIntakeSlots(TEST_USER_ID);
|
||||
expect(summary.slotsCollapsed).toBe(0);
|
||||
expect(summary.rowsSoftDeleted).toBe(0);
|
||||
|
||||
const live = await prisma.medicationIntakeEvent.findMany({
|
||||
where: { userId: TEST_USER_ID, medicationId: med.id, deletedAt: null },
|
||||
});
|
||||
expect(live.map((r) => r.id).sort()).toEqual([a.id, b.id].sort());
|
||||
});
|
||||
|
||||
it("is a no-op on a second run (idempotent)", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const medId = await createTwiceDailyMed();
|
||||
const slot0700 = localHmAsUtc(DAY, "Europe/Berlin", 7, 0);
|
||||
|
||||
await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
scheduledFor: slot0700,
|
||||
takenAt: null,
|
||||
skipped: false,
|
||||
source: "REMINDER",
|
||||
},
|
||||
});
|
||||
await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
scheduledFor: new Date(slot0700.getTime() + 60_000),
|
||||
takenAt: new Date(slot0700.getTime() + 60_000),
|
||||
skipped: false,
|
||||
source: "API",
|
||||
},
|
||||
});
|
||||
|
||||
const first = await dedupeUserIntakeSlots(TEST_USER_ID);
|
||||
expect(first.slotsCollapsed).toBe(1);
|
||||
|
||||
const second = await dedupeUserIntakeSlots(TEST_USER_ID);
|
||||
expect(second.slotsCollapsed).toBe(0);
|
||||
expect(second.rowsSoftDeleted).toBe(0);
|
||||
expect(second.rowsNormalised).toBe(0);
|
||||
|
||||
const live = await prisma.medicationIntakeEvent.findMany({
|
||||
where: { userId: TEST_USER_ID, medicationId: medId, deletedAt: null },
|
||||
});
|
||||
expect(live).toHaveLength(1);
|
||||
expect(live[0]?.takenAt).not.toBeNull();
|
||||
});
|
||||
|
||||
it("keeps the taken row over a skipped row in the same slot", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const medId = await createTwiceDailyMed();
|
||||
const slot0700 = localHmAsUtc(DAY, "Europe/Berlin", 7, 0);
|
||||
|
||||
const skipped = await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
scheduledFor: slot0700,
|
||||
takenAt: null,
|
||||
skipped: true,
|
||||
source: "WEB",
|
||||
},
|
||||
});
|
||||
const taken = await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
medicationId: medId,
|
||||
scheduledFor: new Date(slot0700.getTime() + 30_000),
|
||||
takenAt: new Date(slot0700.getTime() + 30_000),
|
||||
skipped: false,
|
||||
source: "API",
|
||||
},
|
||||
});
|
||||
|
||||
await dedupeUserIntakeSlots(TEST_USER_ID);
|
||||
|
||||
const live = await prisma.medicationIntakeEvent.findMany({
|
||||
where: { userId: TEST_USER_ID, medicationId: medId, deletedAt: null },
|
||||
});
|
||||
expect(live).toHaveLength(1);
|
||||
expect(live[0]?.id).toBe(taken.id);
|
||||
|
||||
const loser = await prisma.medicationIntakeEvent.findUniqueOrThrow({
|
||||
where: { id: skipped.id },
|
||||
});
|
||||
expect(loser.deletedAt).not.toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("intake-slot-dedup — boot discovery scoping", () => {
|
||||
it("returns 0 when no user holds a duplicate pair", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const med = await prisma.medication.create({
|
||||
data: {
|
||||
userId: OTHER_USER_ID,
|
||||
name: "Clean med",
|
||||
dose: "1mg",
|
||||
active: true,
|
||||
schedules: {
|
||||
create: {
|
||||
windowStart: "08:00",
|
||||
windowEnd: "08:00",
|
||||
timesOfDay: ["08:00"],
|
||||
scheduleType: "SCHEDULED",
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
// A single clean row, no duplicate.
|
||||
await prisma.medicationIntakeEvent.create({
|
||||
data: {
|
||||
userId: OTHER_USER_ID,
|
||||
medicationId: med.id,
|
||||
scheduledFor: localHmAsUtc(DAY, "Europe/Berlin", 8, 0),
|
||||
takenAt: null,
|
||||
skipped: false,
|
||||
source: "REMINDER",
|
||||
},
|
||||
});
|
||||
|
||||
// boss is not initialised in this unit context → enqueue helper
|
||||
// returns the no-boss zero result. We assert the discovery query path
|
||||
// by running the per-user pass directly below; the helper's
|
||||
// discovery SQL is exercised in the live worker boot.
|
||||
const result = await enqueueBootTimeIntakeSlotDedup();
|
||||
expect(result.error).toBeNull();
|
||||
expect(result.enqueued).toBe(0);
|
||||
|
||||
// Per-user pass on the clean user is a no-op.
|
||||
const summary = await dedupeUserIntakeSlots(OTHER_USER_ID);
|
||||
expect(summary.slotsCollapsed).toBe(0);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user