diff --git a/src/__tests__/account-deletion-enumeration-guard.test.ts b/src/__tests__/account-deletion-enumeration-guard.test.ts index 2d8026647..cf0bfd600 100644 --- a/src/__tests__/account-deletion-enumeration-guard.test.ts +++ b/src/__tests__/account-deletion-enumeration-guard.test.ts @@ -8,15 +8,20 @@ * that reads across accounts has to filter them out — and a new query is the * one most likely to forget. * - * The rule, for every `findMany(` / `groupBy(` and every raw SQL query that - * reads accounts' rows in the background work below: + * The rule, for every `findMany(` / `groupBy(` / `count(` / `aggregate(`, + * every `updateMany(` / `deleteMany(` (a sweep that writes into every + * account's rows acts for each of them as much as a read does), and every raw + * SQL query or statement that touches accounts' rows in the background work + * below: * * - it names the shared predicate (`ACTIVE_ACCOUNT`, * `OWNED_BY_ACTIVE_ACCOUNT` or `activeAccountSql`), or its result goes * straight into `dropPendingDeletion` (a scan of `measurements`, too * large to join `users` into per row), or * - it is scoped to named accounts (a `userId` it was handed, an `id` list - * it already filtered), or + * it already filtered), read from its filter only: `orderBy`, `by`, + * `distinct`, `select` and `include` name columns without scoping + * anything, or * - a comment directly above it says why it needn't: * `// account-deletion: ` (a ledger read, the purge's own query, * …), or the module says so for all of its queries: @@ -52,18 +57,34 @@ const EXTRA_FILES = [ // Read by the hourly readiness evaluation, which judges every account's // copies. "lib/recovery/readiness.ts", + // Boot discovery and cron sweeps that live beside their domain rather + // than under `lib/jobs`. + "lib/rollups/measurement-rollups.ts", + "lib/rollups/mood-rollups.ts", + "lib/personal-records/cumulative-pr-rederivation.ts", + "lib/medications/intake-slot-dedup.ts", + "lib/medications/inventory/service.ts", + "lib/workouts/climb-backfill.ts", ]; +/** Job folders that live beside their domain. */ +const EXTRA_DIRS = [{ dir: "lib/fitbit/jobs", floor: 5 }]; + function backgroundFiles(): string[] { - const jobs = walkSourceFiles(join(SRC, "lib/jobs"), { floor: 150 }) - .filter((rel) => !rel.includes("__tests__")) - .filter((rel) => !rel.endsWith(".test.ts")) - .map((rel) => `lib/jobs/${rel}`); - return [...jobs, ...EXTRA_FILES].sort(); + const walk = (dir: string, floor: number) => + walkSourceFiles(join(SRC, dir), { floor }) + .filter((rel) => !rel.includes("__tests__")) + .filter((rel) => !rel.endsWith(".test.ts")) + .map((rel) => `${dir}/${rel}`); + return [ + ...walk("lib/jobs", 150), + ...EXTRA_DIRS.flatMap(({ dir, floor }) => walk(dir, floor)), + ...EXTRA_FILES, + ].sort(); } const PREDICATE = - /\b(?:ACTIVE_ACCOUNT|OWNED_BY_ACTIVE_ACCOUNT|activeAccountSql|activeAccount|ownedByActiveAccount)\b/; + /\b(?:ACTIVE_ACCOUNT|OWNED_BY_ACTIVE_ACCOUNT|activeAccountSql|activeAccount|ownedByActiveAccount|excludePendingOwners)\b/; /** A query scoped to accounts it was handed, not one that finds them. */ const SCOPED = [ @@ -71,6 +92,8 @@ const SCOPED = [ // (`userId: true`) or a negation that reads every other account. /\b(?:userId|recordUserId|subjectId|grantorId|profileId)\s*(?:[,}\n]|:(?!\s*(?:true|false)\b|\s*\{\s*(?:not|notIn)\b))/, /\bid\s*:\s*(?:userId|input\.userId|\{\s*in\b)/, + // `id: user.id`, `id: account.id`: the one account a loop is on. + /\bid\s*:\s*[\w$]+\.(?:id|userId)\b/, /"?user_id"?\s*=\s*(?:\$\{|ANY\s*\(\s*\$\{)/, /\bu\."id"\s*=\s*\$\{/, /"id"\s*=\s*(?:\$\{|ANY\s*\(\s*\$\{)/, @@ -142,14 +165,16 @@ function templateEnd(code: string, start: number): number { return i; } -/** Every `findMany(` / `groupBy(` call and account SQL in `source`. */ +/** Every query or sweep call and account SQL in `source`. */ export function queriesIn(file: string, source: string): Query[] { const code = stripComments(source); const lineOf = (index: number) => code.slice(0, index).split("\n").length; const out: Query[] = []; const dropped = (index: number) => DROPPED.test(code.slice(Math.max(0, index - 200), index)); - for (const m of code.matchAll(/\.(?:findMany|groupBy)\s*\(/g)) { + for (const m of code.matchAll( + /\.(?:findMany|groupBy|updateMany|deleteMany|count|aggregate)\s*\(/g, + )) { const open = m.index + m[0].length - 1; out.push({ file, @@ -158,7 +183,9 @@ export function queriesIn(file: string, source: string): Query[] { dropped: dropped(m.index), }); } - for (const m of code.matchAll(/\.\$(?:queryRaw|queryRawUnsafe)\b/g)) { + for (const m of code.matchAll( + /\.\$(?:queryRaw|queryRawUnsafe|executeRaw|executeRawUnsafe)\b/g, + )) { const rest = code.slice(m.index + m[0].length); const tick = rest.search(/[`(]/); if (tick < 0) continue; @@ -182,13 +209,34 @@ function markedAt(source: string, line: number): boolean { return lines.slice(from, line).some((text) => MARKER.test(text)); } +/** + * The parts of a query's arguments that say which rows it reads, without the + * ones that only shape the answer. `orderBy: [{ day: "asc" }, { userId: "asc" }]` + * names `userId` and scopes nothing; read as a scope, it let three nightly + * score jobs group every account's rows, frozen ones included, past this + * guard. The same goes for `by`, `distinct`, `select` and `include`. + */ +const SHAPE_KEY = /\b(?:orderBy|by|distinct|select|include)\s*:\s*(?=[[{])/g; + +export function filterText(text: string): string { + let out = ""; + let from = 0; + for (const m of text.matchAll(SHAPE_KEY)) { + if (m.index < from) continue; + const open = m.index + m[0].length; + out += text.slice(from, m.index); + from = open + balanced(text, open).length; + } + return out + text.slice(from); +} + export function violations(file: string, source: string): Query[] { if (FILE_MARKER.test(source)) return []; return queriesIn(file, source).filter( (q) => !q.dropped && !PREDICATE.test(q.text) && - !SCOPED.some((re) => re.test(q.text)) && + !SCOPED.some((re) => re.test(filterText(q.text))) && !markedAt(source, q.line), ); } @@ -251,4 +299,36 @@ describe("background work skips accounts pending deletion", () => { // Selecting `userId` is not a scope. expect(violations("planted.ts", select)).toHaveLength(2); }); + + // `orderBy: { userId: "asc" }` names the column and scopes nothing; read + // as a scope it hid three nightly score jobs that grouped every account's + // rows, frozen ones included. + it("does not read an order, a grouping or a selection as a scope", () => { + for (const shape of [ + 'orderBy: { userId: "asc" }', + 'orderBy: [{ _max: { measuredAt: "desc" } }, { userId: "asc" }]', + 'by: ["userId"]', + 'distinct: ["userId"]', + "select: { userId: true, user: { select: { id: true } } }", + ]) { + const planted = `export async function s() { await prisma.measurement.groupBy({ where: { type: "WEIGHT" }, ${shape} }); }`; + expect(violations("planted.ts", planted), shape).toHaveLength(1); + } + const scoped = + 'export async function s(userId: string) { await prisma.measurement.findMany({ where: { userId }, orderBy: { userId: "asc" } }); }'; + expect(violations("planted.ts", scoped)).toHaveLength(0); + }); + + it("catches a sweep that writes into every account's rows", () => { + const planted = [ + "export async function sweep() {", + " await prisma.coachPlan.updateMany({", + ' where: { status: "proposed" },', + ' data: { status: "abandoned" },', + " });", + " await prisma.$executeRaw`UPDATE measurements SET x = 1 WHERE user_id IS NOT NULL`;", + "}", + ].join("\n"); + expect(violations("planted.ts", planted)).toHaveLength(2); + }); }); diff --git a/src/lib/account-deletion/active-account.ts b/src/lib/account-deletion/active-account.ts index b50e724bb..b95252fdc 100644 --- a/src/lib/account-deletion/active-account.ts +++ b/src/lib/account-deletion/active-account.ts @@ -63,6 +63,29 @@ export function activeAccountSql(column: string): Prisma.Sql { ); } +/** + * `where` on a table too large to join `users` into row by row + * (`measurements`, `workouts`): its owner is none of the few accounts pending + * deletion, as `userId NOT IN (…)`. Unlike `dropPendingDeletion` it filters + * before a `take`, so a pending account does not use a slot of a capped scan. + */ +export async function excludePendingOwners(db: { + user: { + findMany(args: { + where: { deletionRequestedAt: { not: null } }; + select: { id: true }; + }): Promise>; + }; +}): Promise<{ userId?: { notIn: string[] } }> { + const pending = await db.user.findMany({ + where: { deletionRequestedAt: { not: null } }, + select: { id: true }, + }); + return pending.length === 0 + ? {} + : { userId: { notIn: pending.map((row) => row.id) } }; +} + /** * Drop the accounts pending deletion from the result of a scan over a table * too large to join `users` into row by row (`measurements`): the scan keeps diff --git a/src/lib/fitbit/jobs/__tests__/backfill.test.ts b/src/lib/fitbit/jobs/__tests__/backfill.test.ts index 6f17a4431..7acdcab9a 100644 --- a/src/lib/fitbit/jobs/__tests__/backfill.test.ts +++ b/src/lib/fitbit/jobs/__tests__/backfill.test.ts @@ -63,7 +63,10 @@ describe("enqueueBootTimeFitbitBackfill — discovery", () => { expect(prismaMock.fitbitConnection.findMany).toHaveBeenCalledWith( expect.objectContaining({ - where: { backfillCompletedAt: null }, + where: { + backfillCompletedAt: null, + user: { deletionRequestedAt: null }, + }, }), ); expect(result.enqueued).toBe(2); diff --git a/src/lib/fitbit/jobs/__tests__/sleep-repair.test.ts b/src/lib/fitbit/jobs/__tests__/sleep-repair.test.ts index 374c6fe43..732130aaa 100644 --- a/src/lib/fitbit/jobs/__tests__/sleep-repair.test.ts +++ b/src/lib/fitbit/jobs/__tests__/sleep-repair.test.ts @@ -95,7 +95,12 @@ describe("enqueueBootTimeFitbitSleepRepair — discovery", () => { const result = await enqueueBootTimeFitbitSleepRepair(); expect(prismaMock.fitbitConnection.findMany).toHaveBeenCalledWith( - expect.objectContaining({ where: { sleepRepairedAt: null } }), + expect.objectContaining({ + where: { + sleepRepairedAt: null, + user: { deletionRequestedAt: null }, + }, + }), ); expect(result.enqueued).toBe(2); expect(bossSend).toHaveBeenCalledWith( diff --git a/src/lib/fitbit/jobs/backfill.ts b/src/lib/fitbit/jobs/backfill.ts index 69d81a05f..0a8faa061 100644 --- a/src/lib/fitbit/jobs/backfill.ts +++ b/src/lib/fitbit/jobs/backfill.ts @@ -22,6 +22,7 @@ import { import { syncUserFitbit } from "@/lib/fitbit/sync"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; import { FITBIT_BACKFILL_QUEUE } from "./queues"; +import { OWNED_BY_ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account"; export interface FitbitBackfillPayload { userId: string; @@ -143,7 +144,7 @@ export async function enqueueBootTimeFitbitBackfill( try { const connections = await prisma.fitbitConnection.findMany({ - where: { backfillCompletedAt: null }, + where: { backfillCompletedAt: null, ...OWNED_BY_ACTIVE_ACCOUNT }, select: { userId: true }, }); diff --git a/src/lib/fitbit/jobs/oauth-state-cleanup.ts b/src/lib/fitbit/jobs/oauth-state-cleanup.ts index 97b1dec8e..1391ba3fe 100644 --- a/src/lib/fitbit/jobs/oauth-state-cleanup.ts +++ b/src/lib/fitbit/jobs/oauth-state-cleanup.ts @@ -10,6 +10,7 @@ * Idempotent: a re-run within the same window matches zero rows the second time * because the first pass deleted everything older than `now()`. */ +// account-deletion(file): expires abandoned OAuth state rows; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export async function cleanupExpiredFitbitOAuthStates( diff --git a/src/lib/fitbit/jobs/sleep-repair.ts b/src/lib/fitbit/jobs/sleep-repair.ts index f4919a803..ef5ab4251 100644 --- a/src/lib/fitbit/jobs/sleep-repair.ts +++ b/src/lib/fitbit/jobs/sleep-repair.ts @@ -47,6 +47,7 @@ import { import { syncUserSleep } from "@/lib/fitbit/sync-sleep"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; import { FITBIT_SLEEP_REPAIR_QUEUE } from "./queues"; +import { OWNED_BY_ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account"; export interface FitbitSleepRepairPayload { userId: string; @@ -164,7 +165,7 @@ export async function enqueueBootTimeFitbitSleepRepair( try { const connections = await prisma.fitbitConnection.findMany({ - where: { sleepRepairedAt: null }, + where: { sleepRepairedAt: null, ...OWNED_BY_ACTIVE_ACCOUNT }, select: { userId: true }, }); diff --git a/src/lib/fitbit/jobs/sync-handlers.ts b/src/lib/fitbit/jobs/sync-handlers.ts index 148c89428..db95dab45 100644 --- a/src/lib/fitbit/jobs/sync-handlers.ts +++ b/src/lib/fitbit/jobs/sync-handlers.ts @@ -14,6 +14,7 @@ import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { cleanupExpiredFitbitOAuthStates } from "./oauth-state-cleanup"; import { getWorkerPrisma } from "@/lib/jobs/reminder/shared"; import { isFitbitRetired } from "@/lib/fitbit/retirement"; +import { OWNED_BY_ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account"; /** * v1.12.0 — Fitbit poll-sync payload. Poll-only (no webhook at launch): the @@ -51,6 +52,7 @@ export async function handleFitbitSync( } if (targets.length === 0) { const connections = await prisma.fitbitConnection.findMany({ + where: OWNED_BY_ACTIVE_ACCOUNT, select: { userId: true }, }); targets.push(...connections); diff --git a/src/lib/integrations/watchdog.ts b/src/lib/integrations/watchdog.ts index a5d890469..ffe4b5701 100644 --- a/src/lib/integrations/watchdog.ts +++ b/src/lib/integrations/watchdog.ts @@ -247,6 +247,7 @@ export async function runIntegrationWatchdog(opts: { incidentsOpened: 0, incidentsClosed: 0, notified: 0, + // account-deletion: whether any watch row exists yet; it reads no account. firstRun: (await prisma.integrationWatch.count()) === 0, }; const userIds = await listWatchedUserIds(); diff --git a/src/lib/jobs/__tests__/coach-reminder-sweep.test.ts b/src/lib/jobs/__tests__/coach-reminder-sweep.test.ts index db9cca43f..b89dd2675 100644 --- a/src/lib/jobs/__tests__/coach-reminder-sweep.test.ts +++ b/src/lib/jobs/__tests__/coach-reminder-sweep.test.ts @@ -116,6 +116,7 @@ describe("runCoachReminderSweep", () => { deletedAt: null, status: "proposed", createdAt: { lt: cutoff }, + user: { deletionRequestedAt: null }, }, data: { status: "abandoned", reviewDate: null }, }); @@ -124,6 +125,7 @@ describe("runCoachReminderSweep", () => { deletedAt: null, source: "proposed", createdAt: { lt: cutoff }, + user: { deletionRequestedAt: null }, }, data: { deletedAt: NOW }, }); diff --git a/src/lib/jobs/__tests__/pending-deletion-gate.test.ts b/src/lib/jobs/__tests__/pending-deletion-gate.test.ts new file mode 100644 index 000000000..83b38d5de --- /dev/null +++ b/src/lib/jobs/__tests__/pending-deletion-gate.test.ts @@ -0,0 +1,79 @@ +/** + * A job queued for an account before its deletion was asked for does not run + * once the account is pending deletion: the binding drops it before the + * handler is even loaded, and the rest of its batch runs. + * + * Mutation that turns this red: bind the loaded handler directly again + * (`bindableHandler` without the gate); the pending account's job reaches the + * handler. + */ +import { describe, expect, it, vi } from "vitest"; +import type { Job, PgBoss } from "pg-boss"; + +const pending = vi.hoisted(() => new Set()); +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + findMany: vi.fn(async ({ where }: { where: { id: { in: string[] } } }) => + where.id.in.filter((id) => pending.has(id)).map((id) => ({ id })), + ), + }, + }, +})); + +const { dropJobsOfPendingAccounts, accountOfJob } = + await import("@/lib/jobs/pending-deletion-gate"); +const { bindableHandler } = await import("@/lib/jobs/registry/register"); +const { prisma } = await import("@/lib/db"); + +const job = (id: string, data: unknown) => ({ id, data }) as Job; + +describe("the pending-deletion gate on queued jobs", () => { + it("reads the account a payload names, and nothing else", () => { + expect(accountOfJob(job("1", { userId: "u1", days: 3 }))).toBe("u1"); + expect(accountOfJob(job("2", { restoreJobId: "r" }))).toBeNull(); + expect(accountOfJob(job("3", null))).toBeNull(); + expect(accountOfJob(job("4", { userId: 42 }))).toBeNull(); + }); + + it("drops the jobs of accounts pending deletion and keeps the rest", async () => { + pending.clear(); + pending.add("frozen"); + const batch = [ + job("a", { userId: "frozen" }), + job("b", { userId: "active" }), + job("c", {}), + ]; + const { live, dropped } = await dropJobsOfPendingAccounts(batch); + expect(live.map((j) => j.id)).toEqual(["b", "c"]); + expect(dropped).toBe(1); + }); + + it("asks nothing for a batch that names no account", async () => { + vi.mocked(prisma.user.findMany).mockClear(); + await dropJobsOfPendingAccounts([job("x", { sweep: true })]); + expect(prisma.user.findMany).not.toHaveBeenCalled(); + }); + + it("keeps a bound handler from running a job queued before the deletion", async () => { + pending.clear(); + pending.add("frozen"); + const handler = vi.fn(async () => ({ ok: true as const, did: {} })); + const load = vi.fn(async () => handler); + const bound = bindableHandler( + { queue: "rollup-rebuild", work: {}, load } as never, + { boss: {} as PgBoss }, + ); + + const outcome = await bound([job("1", { userId: "frozen" })]); + expect(outcome).toMatchObject({ + ok: true, + did: { jobs_skipped_deletion_pending: 1 }, + }); + expect(load).not.toHaveBeenCalled(); + + await bound([job("2", { userId: "frozen" }), job("3", { userId: "u" })]); + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledWith([job("3", { userId: "u" })]); + }); +}); diff --git a/src/lib/jobs/__tests__/recovery-score.test.ts b/src/lib/jobs/__tests__/recovery-score.test.ts index ac2b423d4..b44136c8b 100644 --- a/src/lib/jobs/__tests__/recovery-score.test.ts +++ b/src/lib/jobs/__tests__/recovery-score.test.ts @@ -25,14 +25,16 @@ import { const NOW = new Date("2026-06-02T08:00:00Z"); -function makePrisma(userIds: string[]) { +function makePrisma(userIds: string[], pending: string[] = []) { const groupBy = vi .fn() .mockResolvedValue(userIds.map((userId) => ({ userId }))); + const findMany = vi.fn().mockResolvedValue(pending.map((id) => ({ id }))); return { - prisma: { measurement: { groupBy } } as unknown as Parameters< - typeof runRecoveryScore - >[0], + prisma: { + measurement: { groupBy }, + user: { findMany }, + } as unknown as Parameters[0], groupBy, }; } @@ -70,6 +72,18 @@ describe("findRecoveryScoreCandidates", () => { { userId: "asc" }, ]); expect(arg.take).toBe(100); + // No account pending deletion: nothing to leave out. + expect(where.userId).toBeUndefined(); + }); + + // The cohort is capped; a frozen account must neither get a score written + // into its record nor take the slot of an active one. + it("leaves accounts pending deletion out before the cap", async () => { + const { prisma, groupBy } = makePrisma(["a"], ["frozen"]); + await findRecoveryScoreCandidates(prisma, NOW, 100); + expect(groupBy.mock.calls[0][0].where.userId).toEqual({ + notIn: ["frozen"], + }); }); }); diff --git a/src/lib/jobs/account-deletion-purge.ts b/src/lib/jobs/account-deletion-purge.ts index 7a6b7fab3..efa8907e9 100644 --- a/src/lib/jobs/account-deletion-purge.ts +++ b/src/lib/jobs/account-deletion-purge.ts @@ -152,6 +152,7 @@ async function deleteExpiredPreWipeCopies(now: Date): Promise { where: { status: { in: ["queued", "running"] } }, select: { backupId: true }, }); + // account-deletion: the purge's own sweep; a copy's window ends whatever its account's state. const { count } = await prisma.dataBackup.deleteMany({ where: { type: { startsWith: PRE_WIPE_BACKUP_TYPE_PREFIX }, diff --git a/src/lib/jobs/coach-message-cleanup.ts b/src/lib/jobs/coach-message-cleanup.ts index 78768d9c1..f1e4469ca 100644 --- a/src/lib/jobs/coach-message-cleanup.ts +++ b/src/lib/jobs/coach-message-cleanup.ts @@ -43,6 +43,7 @@ export async function cleanupOldCoachMessages( ): Promise { const days = getCoachMessageRetentionDays(); const cutoff = new Date(now.getTime() - days * 86_400_000); + // account-deletion: retention ages rows out whatever their account's state; a frozen record keeps no more than an active one. const { count } = await prisma.coachMessage.deleteMany({ where: { createdAt: { lt: cutoff } }, }); diff --git a/src/lib/jobs/coach-reminder-sweep.ts b/src/lib/jobs/coach-reminder-sweep.ts index 2bc4bcc30..f4395f37a 100644 --- a/src/lib/jobs/coach-reminder-sweep.ts +++ b/src/lib/jobs/coach-reminder-sweep.ts @@ -154,6 +154,7 @@ export async function runCoachReminderSweep( deletedAt: null, status: "proposed", createdAt: { lt: lapseBefore }, + ...OWNED_BY_ACTIVE_ACCOUNT, }, data: { status: "abandoned", reviewDate: null }, }); @@ -163,6 +164,7 @@ export async function runCoachReminderSweep( deletedAt: null, source: PROPOSED_FACT_SOURCE, createdAt: { lt: lapseBefore }, + ...OWNED_BY_ACTIVE_ACCOUNT, }, data: { deletedAt: now }, }); @@ -336,6 +338,7 @@ export async function runCoachReminderSweep( status: "surfaced", lastSurfacedAt: { lte: new Date(now.getTime() - NAG_STALE_MS) }, ...notHeld, + ...OWNED_BY_ACTIVE_ACCOUNT, }, data: { surfaceCount: { increment: 1 } }, }); @@ -345,6 +348,7 @@ export async function runCoachReminderSweep( status: "surfaced", surfaceCount: { gte: REMINDER_NAG_CAP }, ...notHeld, + ...OWNED_BY_ACTIVE_ACCOUNT, }, data: { status: "dismissed" }, }); diff --git a/src/lib/jobs/data-arrival.ts b/src/lib/jobs/data-arrival.ts index 499689940..6f3a81f6b 100644 --- a/src/lib/jobs/data-arrival.ts +++ b/src/lib/jobs/data-arrival.ts @@ -153,6 +153,7 @@ async function claimReaction( `; } + // account-deletion: one row of the account this arrival belongs to, by id. const advanced = await tx.arrivalReaction.updateMany({ where: { id: current.id, occurredAt: { lt: occurredAt } }, data: { diff --git a/src/lib/jobs/document-ai-run.ts b/src/lib/jobs/document-ai-run.ts index 4df270a6a..d3eae75c8 100644 --- a/src/lib/jobs/document-ai-run.ts +++ b/src/lib/jobs/document-ai-run.ts @@ -179,6 +179,7 @@ async function armDeadline( "The AI response time is set longer than a background read may run.", }); } + // account-deletion: the one run this job claimed, by id. await prisma.documentAiRun.updateMany({ where: { id: runId, status: "RUNNING" }, data: { expiresAt: new Date(ends) }, diff --git a/src/lib/jobs/document-summary-reaper.ts b/src/lib/jobs/document-summary-reaper.ts index 775a90926..4069ebac5 100644 --- a/src/lib/jobs/document-summary-reaper.ts +++ b/src/lib/jobs/document-summary-reaper.ts @@ -45,6 +45,7 @@ export async function reapStalePendingSummaries( now: Date = new Date(), ): Promise { const cutoff = new Date(now.getTime() - SUMMARY_PENDING_TTL_MS); + // account-deletion: releases summaries stuck in PENDING by a run that died; it starts no work. const { count } = await prisma.inboundDocument.updateMany({ where: { summaryState: "PENDING", updatedAt: { lt: cutoff } }, data: { summaryState: "UNAVAILABLE" }, diff --git a/src/lib/jobs/document-thumbnail.ts b/src/lib/jobs/document-thumbnail.ts index 4195820ee..91d0f2f7d 100644 --- a/src/lib/jobs/document-thumbnail.ts +++ b/src/lib/jobs/document-thumbnail.ts @@ -151,6 +151,7 @@ async function renderThumbnail( }, }); + // account-deletion: the one document this job was handed. await prisma.documentThumbnailFailure.deleteMany({ where: { documentId: document.id }, }); diff --git a/src/lib/jobs/google-health-oauth-state-cleanup.ts b/src/lib/jobs/google-health-oauth-state-cleanup.ts index 7f807b379..d41787b71 100644 --- a/src/lib/jobs/google-health-oauth-state-cleanup.ts +++ b/src/lib/jobs/google-health-oauth-state-cleanup.ts @@ -10,6 +10,7 @@ * Idempotent: a re-run within the same window matches zero rows the second time * because the first pass deleted everything older than `now()`. */ +// account-deletion(file): expires abandoned OAuth state rows; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export async function cleanupExpiredGoogleHealthOAuthStates( diff --git a/src/lib/jobs/host-metric-sampler.ts b/src/lib/jobs/host-metric-sampler.ts index 9a93f63cc..107f5f4b8 100644 --- a/src/lib/jobs/host-metric-sampler.ts +++ b/src/lib/jobs/host-metric-sampler.ts @@ -22,6 +22,7 @@ * tick. With one insert every five minutes that's 2,016 rows steady-state, * indexed by `captured_at` so the 2h query stays under 30 rows. */ +// account-deletion(file): host metrics belong to no account. import { loadavg, totalmem, freemem } from "node:os"; import { readFile } from "node:fs/promises"; import type { PrismaClient } from "@/generated/prisma/client"; diff --git a/src/lib/jobs/idempotency-cleanup.ts b/src/lib/jobs/idempotency-cleanup.ts index ec30f8fb5..d0ba74d7e 100644 --- a/src/lib/jobs/idempotency-cleanup.ts +++ b/src/lib/jobs/idempotency-cleanup.ts @@ -6,6 +6,7 @@ * — most rows never see another retry, so without a sweeper the table * grows unbounded with stale 24h-expired entries. */ +// account-deletion(file): expires cached request replays; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export async function cleanupExpiredIdempotencyKeys( diff --git a/src/lib/jobs/intake-auto-skip.ts b/src/lib/jobs/intake-auto-skip.ts index 4cb6cdbfd..780cf27df 100644 --- a/src/lib/jobs/intake-auto-skip.ts +++ b/src/lib/jobs/intake-auto-skip.ts @@ -318,6 +318,7 @@ export async function runIntakeAutoSkipPass( const invalidatedUserIds = new Set(); for (const [delayMs, group] of medsByDelay) { const medicationIds = group.map((m) => m.id); + // account-deletion: the medications of active accounts read above (`medicationIds`). const { count } = await prisma.medicationIntakeEvent.updateMany({ where: { ...pendingWhere, diff --git a/src/lib/jobs/intake-auto-take.ts b/src/lib/jobs/intake-auto-take.ts index 1e265a750..23ef5e980 100644 --- a/src/lib/jobs/intake-auto-take.ts +++ b/src/lib/jobs/intake-auto-take.ts @@ -527,6 +527,7 @@ async function applyDecision( if (decision.kind === "convert") { // Conditional: still pending at the moment of the write. A take or skip // the person records concurrently makes this match nothing. + // account-deletion: the one row this decision is about, by id. const { count } = await prisma.medicationIntakeEvent.updateMany({ where: { id: decision.rowId, diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 46d022cf3..3ede654d1 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -45,8 +45,12 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "access_tokens_deleted", // The account-deletion purge: accounts whose grace window ended and // were erased, ones whose erasure was refused and stays pending, and the - // copies taken before a data wipe that aged out with their window. + // copies taken before a data wipe that aged out with their window; the + // erasures that threw, and the due accounts left for the next run when + // its time budget ran out. + "account_deletions_deferred", "account_deletions_due", + "account_deletions_failed", "account_deletions_purged", "account_deletions_refused", "accounts_completed", @@ -160,6 +164,9 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "inventory_expired_count", "inventory_items_migrated", "jobs", + // A batch whose every job named an account pending deletion + // (`pending-deletion-gate.ts`): completed without running. + "jobs_skipped_deletion_pending", // Heavy jobs that found the lane taken and went back to their queue. "lane_deferred", "legacy_rows_soft_deleted", diff --git a/src/lib/jobs/mcp-token-cleanup.ts b/src/lib/jobs/mcp-token-cleanup.ts index d5d5869a9..20a76913b 100644 --- a/src/lib/jobs/mcp-token-cleanup.ts +++ b/src/lib/jobs/mcp-token-cleanup.ts @@ -13,6 +13,7 @@ * revoked connection anchors are pruned too once no access token references * them, keeping the `mcp_oauth_connections` table bounded. */ +// account-deletion(file): removes revoked or expired connector tokens; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; import { SCOPE_HEALTH_READ } from "@/lib/mcp/oauth/config"; diff --git a/src/lib/jobs/measurement-reminder.ts b/src/lib/jobs/measurement-reminder.ts index e81f56bfa..b8997bb2e 100644 --- a/src/lib/jobs/measurement-reminder.ts +++ b/src/lib/jobs/measurement-reminder.ts @@ -165,12 +165,12 @@ async function cleanupExpiredCoachReminders( now: Date, ): Promise { const result = await prisma.measurementReminder.updateMany({ - where: { + where: ownedByActiveAccount({ deletedAt: null, - origin: "COACH", + origin: "COACH" as const, endsOn: { not: null, lt: now }, nextDueAt: null, - }, + }), data: { deletedAt: now }, }); return result.count; diff --git a/src/lib/jobs/medication-intake-import.ts b/src/lib/jobs/medication-intake-import.ts index bea85b8c1..2e91053be 100644 --- a/src/lib/jobs/medication-intake-import.ts +++ b/src/lib/jobs/medication-intake-import.ts @@ -765,6 +765,7 @@ async function recordWorkerFailure( const retryLimit = metadata.retryLimit ?? 0; const terminal = retryCount >= retryLimit; const now = new Date(); + // account-deletion: the one import this job runs, by id. await client.medicationIntakeImportJob.updateMany({ where: { id: job.data.jobId, diff --git a/src/lib/jobs/oauth-callback-claim-cleanup.ts b/src/lib/jobs/oauth-callback-claim-cleanup.ts index cd2ba7a5a..c21c3bedf 100644 --- a/src/lib/jobs/oauth-callback-claim-cleanup.ts +++ b/src/lib/jobs/oauth-callback-claim-cleanup.ts @@ -8,6 +8,7 @@ * This pass rides the daily Withings OAuth-state cleanup (03:20 Europe/Berlin) * so the table empties on its own regardless. Idempotent. */ +// account-deletion(file): expires single-use callback claims; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export async function cleanupExpiredOAuthCallbackClaims( diff --git a/src/lib/jobs/offhost-backup.ts b/src/lib/jobs/offhost-backup.ts index 36fe30ce6..9899c83c2 100644 --- a/src/lib/jobs/offhost-backup.ts +++ b/src/lib/jobs/offhost-backup.ts @@ -933,6 +933,7 @@ export async function runOffhostBackup( // The account was deleted or wiped while its copy was being written. // The purge that request started may already have run, so the copy // this run just put there is removed here rather than left behind. + // account-deletion: the one account this upload is for. const purgedMeanwhile = (await prisma.offhostPurgeRequest.count({ where: { diff --git a/src/lib/jobs/oidc-handoff-cleanup.ts b/src/lib/jobs/oidc-handoff-cleanup.ts index a6fb6e18a..0777592fd 100644 --- a/src/lib/jobs/oidc-handoff-cleanup.ts +++ b/src/lib/jobs/oidc-handoff-cleanup.ts @@ -11,6 +11,7 @@ * Idempotent: a re-run within the same window matches zero rows the second time * because the first pass deleted everything older than `now()`. */ +// account-deletion(file): expires single-use sign-in handoffs; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export async function cleanupExpiredOidcNativeHandoffs( diff --git a/src/lib/jobs/pending-deletion-gate.ts b/src/lib/jobs/pending-deletion-gate.ts new file mode 100644 index 000000000..0c9ea6dea --- /dev/null +++ b/src/lib/jobs/pending-deletion-gate.ts @@ -0,0 +1,58 @@ +/** + * The chokepoint every queued job for one account passes before it runs. + * + * Background passes that enumerate accounts skip one pending deletion + * (`ACTIVE_ACCOUNT`, held by the enumeration guard). A job queued for an + * account before its deletion was asked for (a provider backfill, an import, + * a rollup rebuild, a document index, a record detection) was decided before + * the account was frozen, and its handler acts on the `userId` it carries + * without asking again. So the binding asks for it: a job whose payload names + * an account pending deletion is completed without running, and the rest of + * its batch runs as before. An account that no longer exists is not this + * gate's business; the handler finds nothing to act on. + */ +import type { Job } from "pg-boss"; + +/** The account a job's payload names, when it names one. */ +export function accountOfJob(job: Pick, "data">): string | null { + const data = job.data as { userId?: unknown } | null | undefined; + return data && typeof data === "object" && typeof data.userId === "string" + ? data.userId + : null; +} + +/** Which of `ids` are pending deletion. */ +export type PendingReader = (ids: string[]) => Promise>; + +const readPending: PendingReader = async (ids) => { + const { prisma } = await import("@/lib/db"); + const rows = await prisma.user.findMany({ + where: { id: { in: ids }, deletionRequestedAt: { not: null } }, + select: { id: true }, + }); + return new Set(rows.map((row) => row.id)); +}; + +/** + * Split a batch into the jobs that may run and the number dropped because + * their account is pending deletion. A batch with no account in it costs no + * query. + */ +export async function dropJobsOfPendingAccounts< + J extends Pick, "data">, +>( + jobs: J[], + pendingOf: PendingReader = readPending, +): Promise<{ live: J[]; dropped: number }> { + const ids = [ + ...new Set(jobs.map(accountOfJob).filter((id) => id !== null)), + ] as string[]; + if (ids.length === 0) return { live: jobs, dropped: 0 }; + const pending = await pendingOf(ids); + if (pending.size === 0) return { live: jobs, dropped: 0 }; + const live = jobs.filter((job) => { + const id = accountOfJob(job); + return id === null || !pending.has(id); + }); + return { live, dropped: jobs.length - live.length }; +} diff --git a/src/lib/jobs/recovery-score.ts b/src/lib/jobs/recovery-score.ts index c0a4a796f..8f354fbf1 100644 --- a/src/lib/jobs/recovery-score.ts +++ b/src/lib/jobs/recovery-score.ts @@ -23,6 +23,7 @@ import type { PrismaClient } from "@/generated/prisma/client"; import { persistRecoveryScore } from "@/lib/insights/recovery-score"; import { runScoreBatch } from "@/lib/insights/score-row"; +import { excludePendingOwners } from "@/lib/account-deletion/active-account"; /** * Per-run user cap. A nightly fan-out across every user could starve the @@ -78,6 +79,7 @@ export async function findRecoveryScoreCandidates( }, deletedAt: null, measuredAt: { gte: since }, + ...(await excludePendingOwners(prisma)), }, orderBy: [{ _max: { measuredAt: "desc" } }, { userId: "asc" }], take: cap, diff --git a/src/lib/jobs/registry/register.ts b/src/lib/jobs/registry/register.ts index f53205381..412495a56 100644 --- a/src/lib/jobs/registry/register.ts +++ b/src/lib/jobs/registry/register.ts @@ -25,6 +25,8 @@ import type { Job, PgBoss } from "pg-boss"; import { heavyLane } from "@/lib/jobs/heavy-lane"; +import { jobDone } from "@/lib/jobs/job-outcome"; +import { dropJobsOfPendingAccounts } from "@/lib/jobs/pending-deletion-gate"; import { lockedPass } from "@/lib/jobs/long-pass"; import { createAndSchedule, @@ -74,15 +76,23 @@ export function handlerLoader( /** * The handler a definition binds: its loaded module behind the lock and the - * lane it declares. + * lane it declares, and behind the gate that drops a job for an account + * pending deletion. */ export function bindableHandler( definition: JobDefinition, ctx: JobContext, load: HandlerLoader = handlerLoader(definition, ctx), ): BatchHandler> { - let handler: BatchHandler> = async (jobs) => - (await load())(jobs); + // A job for an account pending deletion completes without running: it + // was queued before the account was frozen (see `pending-deletion-gate`). + let handler: BatchHandler> = async (jobs) => { + const { live, dropped } = await dropJobsOfPendingAccounts(jobs); + if (live.length === 0) { + return jobDone({ jobs_skipped_deletion_pending: dropped }); + } + return (await load())(live); + }; if (definition.lane) { handler = heavyLane( definition.queue, diff --git a/src/lib/jobs/reminder/cleanup-handlers.ts b/src/lib/jobs/reminder/cleanup-handlers.ts index b5d28bedf..c1a4193f1 100644 --- a/src/lib/jobs/reminder/cleanup-handlers.ts +++ b/src/lib/jobs/reminder/cleanup-handlers.ts @@ -108,6 +108,7 @@ export async function handleMoodReminderCleanup( cutoff.setUTCDate(cutoff.getUTCDate() - MOOD_REMINDER_RETENTION_DAYS); // eslint-disable-next-line healthlog/no-utc-day-key -- UTC by design: retention cutoff; a day either way does not matter const cutoffIso = cutoff.toISOString().slice(0, 10); + // account-deletion: retention ages dispatch rows out whatever their account's state. const deleted = await p.moodReminderDispatch.deleteMany({ where: { date: { lt: cutoffIso } }, }); @@ -156,15 +157,18 @@ export async function handlePushAttemptCleanup( try { const cutoff = new Date(); cutoff.setUTCDate(cutoff.getUTCDate() - PUSH_ATTEMPT_RETENTION_DAYS); + // account-deletion: retention ages delivery ledger rows out whatever their account's state. const deletedAttempts = await p.pushAttempt.deleteMany({ where: { createdAt: { lt: cutoff } }, }); + // account-deletion: retention, as above. const deletedEvents = await p.notificationEvent.deleteMany({ where: { createdAt: { lt: cutoff }, eventType: { not: CONFIG_NOTICE_EVENT_TYPE }, }, }); + // account-deletion: retention, as above. const deletedAuthorizations = await p.notificationEgressAuthorization.deleteMany({ where: { authorizedAt: { lt: cutoff } }, @@ -211,6 +215,7 @@ export async function handleArrivalReactionCleanup( try { const cutoff = new Date(); cutoff.setUTCDate(cutoff.getUTCDate() - ARRIVAL_REACTION_RETENTION_DAYS); + // account-deletion: retention ages rows out whatever their account's state. const deleted = await p.arrivalReaction.deleteMany({ where: { createdAt: { lt: cutoff } }, }); @@ -343,6 +348,7 @@ export async function handleStepUpElevationCleanup( return withBackgroundEvent("job.step_up_elevation_cleanup", async (evt) => { const p = getWorkerPrisma(); try { + // account-deletion: expires single-use step-up proofs; acts for no account. const { count } = await p.stepUpElevation.deleteMany({ where: { expiresAt: { lt: new Date() } }, }); diff --git a/src/lib/jobs/reminder/medication-reminder-check.ts b/src/lib/jobs/reminder/medication-reminder-check.ts index 2dd0529a1..2c9e85e5b 100644 --- a/src/lib/jobs/reminder/medication-reminder-check.ts +++ b/src/lib/jobs/reminder/medication-reminder-check.ts @@ -160,7 +160,7 @@ export async function handleReminderCheck( // Clean up expired snoozes await prisma.medication.updateMany({ - where: { snoozedUntil: { lt: now } }, + where: { snoozedUntil: { lt: now }, ...OWNED_BY_ACTIVE_ACCOUNT }, data: { snoozedUntil: null }, }); diff --git a/src/lib/jobs/reminder/telegram-deletion-cleanup.ts b/src/lib/jobs/reminder/telegram-deletion-cleanup.ts index 4088ddb62..6ace5efcb 100644 --- a/src/lib/jobs/reminder/telegram-deletion-cleanup.ts +++ b/src/lib/jobs/reminder/telegram-deletion-cleanup.ts @@ -83,6 +83,7 @@ export async function cleanupScheduledTelegramDeletions(): Promise { // TelegramPromptContext row; once past `expiresAt` it can never bind a // reply, so drop it. The prompt MESSAGE itself self-cleans through the // scheduled-deletion rows above. + // account-deletion: expires reply contexts that can no longer bind; acts for no account. const prunedContexts = await prisma.telegramPromptContext.deleteMany({ where: { expiresAt: { lte: new Date() } }, }); diff --git a/src/lib/jobs/strain-score.ts b/src/lib/jobs/strain-score.ts index afe09ddab..ce8192d46 100644 --- a/src/lib/jobs/strain-score.ts +++ b/src/lib/jobs/strain-score.ts @@ -20,6 +20,7 @@ import type { PrismaClient } from "@/generated/prisma/client"; import { persistStrainScore } from "@/lib/insights/strain-score"; import { runScoreBatch } from "@/lib/insights/score-row"; +import { excludePendingOwners } from "@/lib/account-deletion/active-account"; /** Per-run user cap — same starvation guard the other score passes use. */ export const STRAIN_SCORE_BATCH_CAP = 500; @@ -55,10 +56,14 @@ export async function findStrainScoreCandidates( // order picked an arbitrary set when more than `cap` users qualified; the // ordered `groupBy` makes each side take the most-recently-active users, and // the merge below is order-stable. + // An account pending deletion gets no score and takes no slot of the cap. const [workoutUsers, energyUsers] = await Promise.all([ prisma.workout.groupBy({ by: ["userId"], - where: { startedAt: { gte: since } }, + where: { + startedAt: { gte: since }, + ...(await excludePendingOwners(prisma)), + }, orderBy: [{ _max: { startedAt: "desc" } }, { userId: "asc" }], take: cap, }), @@ -68,6 +73,7 @@ export async function findStrainScoreCandidates( type: "ACTIVE_ENERGY_BURNED", deletedAt: null, measuredAt: { gte: since }, + ...(await excludePendingOwners(prisma)), }, orderBy: [{ _max: { measuredAt: "desc" } }, { userId: "asc" }], take: cap, diff --git a/src/lib/jobs/stress-score.ts b/src/lib/jobs/stress-score.ts index 8a73866bf..72b104b1f 100644 --- a/src/lib/jobs/stress-score.ts +++ b/src/lib/jobs/stress-score.ts @@ -21,6 +21,7 @@ import type { PrismaClient } from "@/generated/prisma/client"; import { persistStressScore } from "@/lib/insights/stress-score"; import { runScoreBatch } from "@/lib/insights/score-row"; +import { excludePendingOwners } from "@/lib/account-deletion/active-account"; /** Per-run user cap — same starvation guard the recovery-score pass uses. */ export const STRESS_SCORE_BATCH_CAP = 500; @@ -61,6 +62,7 @@ export async function findStressScoreCandidates( type: "HEART_RATE_VARIABILITY", deletedAt: null, measuredAt: { gte: since }, + ...(await excludePendingOwners(prisma)), }, orderBy: [{ _max: { measuredAt: "desc" } }, { userId: "asc" }], take: cap, diff --git a/src/lib/jobs/whoop-oauth-state-cleanup.ts b/src/lib/jobs/whoop-oauth-state-cleanup.ts index 59ca863d1..6b8c18237 100644 --- a/src/lib/jobs/whoop-oauth-state-cleanup.ts +++ b/src/lib/jobs/whoop-oauth-state-cleanup.ts @@ -10,6 +10,7 @@ * Idempotent: a re-run within the same window matches zero rows the second time * because the first pass deleted everything older than `now()`. */ +// account-deletion(file): expires abandoned OAuth state rows and connect tickets; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export async function cleanupExpiredWhoopOAuthStates( diff --git a/src/lib/jobs/withings-oauth-state-cleanup.ts b/src/lib/jobs/withings-oauth-state-cleanup.ts index 2cf9718c8..318f7e6f3 100644 --- a/src/lib/jobs/withings-oauth-state-cleanup.ts +++ b/src/lib/jobs/withings-oauth-state-cleanup.ts @@ -18,6 +18,7 @@ * matches zero rows the second time because the first pass deleted * everything older than `now()`. */ +// account-deletion(file): expires abandoned OAuth state rows; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export async function cleanupExpiredWithingsOAuthStates( diff --git a/src/lib/jobs/workout-insight-claim-cleanup.ts b/src/lib/jobs/workout-insight-claim-cleanup.ts index b51bd799b..77b7bddb4 100644 --- a/src/lib/jobs/workout-insight-claim-cleanup.ts +++ b/src/lib/jobs/workout-insight-claim-cleanup.ts @@ -12,6 +12,7 @@ * its `local_date` says, so no row that the cap can still read is ever * deleted. */ +// account-deletion(file): expires generation claims; acts for no account. import type { PrismaClient } from "@/generated/prisma/client"; export const WORKOUT_INSIGHT_CLAIM_RETENTION_DAYS = 7; diff --git a/src/lib/medications/intake-slot-dedup.ts b/src/lib/medications/intake-slot-dedup.ts index 982436f60..c1828e5d4 100644 --- a/src/lib/medications/intake-slot-dedup.ts +++ b/src/lib/medications/intake-slot-dedup.ts @@ -64,6 +64,7 @@ import type { WorkerScheduleRow, } from "@/lib/medications/scheduling/worker-helpers"; import { refreshDoseSlot } from "@/lib/medications/dose-day-refresh"; +import { activeAccountSql } from "@/lib/account-deletion/active-account"; export const INTAKE_SLOT_DEDUP_QUEUE = "intake-slot-dedup"; @@ -614,6 +615,7 @@ async function convergeAdhocRowsOntoPendingSlots(input: { if (pending.takenAt !== null || pending.skipped) continue; // already served. // Tombstone the pending row; the take is the dose of record. + // account-deletion: one row of the account this pass was handed, by id. await prisma.medicationIntakeEvent.updateMany({ where: { id: pending.id, deletedAt: null }, data: { deletedAt: new Date(), syncVersion: { increment: 1 } }, @@ -737,6 +739,7 @@ export async function enqueueBootTimeIntakeSlotDedup(): Promise<{ AND abs(extract(epoch FROM (a."scheduled_for" - b."scheduled_for"))) <= 475200 ) ) + WHERE ${activeAccountSql('a."user_id"')} `; if (users.length === 0) { diff --git a/src/lib/medications/inventory/__tests__/service.test.ts b/src/lib/medications/inventory/__tests__/service.test.ts index bb841c2ae..00832cb6c 100644 --- a/src/lib/medications/inventory/__tests__/service.test.ts +++ b/src/lib/medications/inventory/__tests__/service.test.ts @@ -41,6 +41,8 @@ describe("expireStaleInUseItems", () => { where: { state: "IN_USE", expiresAt: { lt: new Date(NOW.getTime()) }, + // The cron leaves an account pending deletion as it was. + user: { deletionRequestedAt: null }, }, data: { state: "EXPIRED" }, }); diff --git a/src/lib/medications/inventory/service.ts b/src/lib/medications/inventory/service.ts index 0744b259d..e6df0e2b2 100644 --- a/src/lib/medications/inventory/service.ts +++ b/src/lib/medications/inventory/service.ts @@ -20,6 +20,7 @@ import { type SupplyItemState, type SupplySummary, } from "./summary"; +import { OWNED_BY_ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account"; /** * Daily cron pass — flip any IN_USE row whose persisted `expiresAt` has @@ -55,7 +56,9 @@ export async function expireStaleInUseItems(input: { where: { state: "IN_USE", expiresAt: { lt: now }, - ...(userId ? { userId } : {}), + // The cron sweeps every account but one pending deletion, whose record + // stays as it was; the per-user trigger names its account. + ...(userId ? { userId } : OWNED_BY_ACTIVE_ACCOUNT), }, data: { state: "EXPIRED" }, }); diff --git a/src/lib/notifications/__tests__/medication-intake-sync.test.ts b/src/lib/notifications/__tests__/medication-intake-sync.test.ts index 043a4bf4a..4a54b3c07 100644 --- a/src/lib/notifications/__tests__/medication-intake-sync.test.ts +++ b/src/lib/notifications/__tests__/medication-intake-sync.test.ts @@ -38,9 +38,13 @@ vi.mock("@/lib/logging/context", () => ({ const deviceFindMany = vi.fn(); const channelFindUnique = vi.fn(); const deviceDeleteMany = vi.fn(); +const userFindUnique = vi.fn(); vi.mock("@/lib/db", () => ({ prisma: { + user: { + findUnique: (...a: unknown[]) => userFindUnique(...a), + }, device: { findMany: (...a: unknown[]) => deviceFindMany(...a), deleteMany: (...a: unknown[]) => deviceDeleteMany(...a), @@ -90,12 +94,40 @@ beforeEach(() => { channelFindUnique.mockResolvedValue({ enabled: true, preferences: [] }); rawSendMock.mockResolvedValue({ ok: true, status: 200 }); deviceDeleteMany.mockResolvedValue({ count: 0 }); + userFindUnique.mockResolvedValue({ + deletionRequestedAt: null, + deletionPurgeAt: null, + }); }); afterEach(() => { vi.useRealTimers(); }); +describe("queueMedicationIntakeSync — an account pending deletion", () => { + // The push does not go through the dispatcher, whose chokepoint refuses a + // pending account; an import queued before the deletion reached it. + it("sends nothing to the devices of an account pending deletion", async () => { + deviceFindMany.mockResolvedValue([device()]); + userFindUnique.mockResolvedValue({ + deletionRequestedAt: new Date("2026-10-20T00:00:00Z"), + deletionPurgeAt: new Date("2026-11-03T00:00:00Z"), + }); + queueMedicationIntakeSync({ + userId: nextUserId(), + originDeviceToken: null, + }); + await drainFlush(); + expect(rawSendMock).not.toHaveBeenCalled(); + expect(recordPushAttemptMock).toHaveBeenCalledWith( + expect.objectContaining({ + result: "skipped", + reason: "deletion_pending", + }), + ); + }); +}); + describe("queueMedicationIntakeSync — origin skip + other-devices-only", () => { it("dispatches immediately to every device EXCEPT the origin", async () => { deviceFindMany.mockResolvedValue([ diff --git a/src/lib/notifications/medication-intake-sync.ts b/src/lib/notifications/medication-intake-sync.ts index 0950fea7d..18e9a17a7 100644 --- a/src/lib/notifications/medication-intake-sync.ts +++ b/src/lib/notifications/medication-intake-sync.ts @@ -57,6 +57,7 @@ * must never fail the intake mutation that triggered it; the canonical * row is already persisted. */ +import { pendingDeletionOf } from "@/lib/account-deletion/pending"; import { prisma } from "@/lib/db"; import { annotate, getEvent } from "@/lib/logging/context"; import { @@ -193,6 +194,19 @@ async function flushIntakeSync( return; } + // The dispatcher refuses every delivery for an account pending deletion; + // this push does not go through it, so it asks the same question. + if (await pendingDeletionOf(userId)) { + recordPushAttempt({ + userId, + channel: "APNS", + eventType: EVENT_TYPE, + result: "skipped", + reason: "deletion_pending", + }); + return; + } + if (!(await apnsSyncEnabled(userId))) { recordPushAttempt({ userId, diff --git a/src/lib/personal-records/cumulative-pr-rederivation.ts b/src/lib/personal-records/cumulative-pr-rederivation.ts index f428bb922..a57533bcd 100644 --- a/src/lib/personal-records/cumulative-pr-rederivation.ts +++ b/src/lib/personal-records/cumulative-pr-rederivation.ts @@ -42,6 +42,7 @@ import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; import { detectPersonalRecordsForUser } from "@/lib/personal-records/pr-detection-worker"; import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import { OWNED_BY_ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account"; /** Budget for one account's delete-and-re-derive transaction. */ const REDERIVE_TRANSACTION_TIMEOUT_MS = 5 * 60 * 1000; @@ -163,6 +164,7 @@ export async function enqueueBootTimeCumulativePrRederivation( metricSlot: null, metricType: { in: CUMULATIVE_TYPES_ARRAY }, createdAt: { lt: CUMULATIVE_PR_FIX_CUTOFF }, + ...OWNED_BY_ACTIVE_ACCOUNT, }, distinct: ["userId"], select: { userId: true }, diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index bf3343d18..200799b83 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -34,6 +34,7 @@ * took. Nothing here reads both tiers; the rollup replaces the legacy read and * falls back to live SQL only on a coverage miss. */ +import { activeAccountSql } from "@/lib/account-deletion/active-account"; import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; @@ -1000,6 +1001,7 @@ async function persistRollupRows( // window is authoritatively empty, so every bucket in it must go. No // transaction needed — there is no insert leg that could leave a gap. if (!pruneWhere) return 0; + // account-deletion: scoped to the one account this fold was handed (`pruneWhere` carries its userId). await prisma.measurementRollup.deleteMany({ where: pruneWhere }); return 0; } @@ -1065,6 +1067,7 @@ async function persistRollupRows( async (tx) => { if (statementTimeoutMs) await applyStatementBudget(tx, statementTimeoutMs); + // account-deletion: scoped to the one account this fold was handed (`deleteWhere` carries its userId). await tx.measurementRollup.deleteMany({ where: deleteWhere }); for (let i = 0; i < ordered.length; i += CHUNK) { const slice = ordered.slice(i, i + CHUNK); @@ -1436,6 +1439,7 @@ export async function enqueueBootTimeRollupBackfill( AND r."bucket_start" >= ${foldWindowStart} AND (r."sum_xy" IS NULL OR r."reading_min_value" IS NULL) ) discovered + WHERE ${activeAccountSql('discovered."id"')} `; if (users.length === 0) { diff --git a/src/lib/rollups/mood-rollups.ts b/src/lib/rollups/mood-rollups.ts index cf249bb05..ef6a8dd14 100644 --- a/src/lib/rollups/mood-rollups.ts +++ b/src/lib/rollups/mood-rollups.ts @@ -45,6 +45,7 @@ * `enqueueBootTimeMoodRollupBackfill` mints the rollups on the * next worker boot so subsequent reads converge on the fast path. */ +import { activeAccountSql } from "@/lib/account-deletion/active-account"; import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; @@ -564,6 +565,7 @@ export async function enqueueBootTimeMoodRollupBackfill( ON r."user_id" = mt."user_id" AND r."granularity" = 'DAY' WHERE r."bucket_start" IS NULL + AND ${activeAccountSql('mt."user_id"')} `; if (users.length === 0) { @@ -685,6 +687,7 @@ async function persistMoodRollupRows( (row) => Prisma.sql`(${userId}, ${granularity}::"measurement_rollup_granularity", ${row.bucket_start}, ${Number(row.count)}, ${row.mean ?? 0}, ${row.min_score ?? 0}, ${row.max_score ?? 0}, ${row.sd}, ${computedAt})`, ); + // account-deletion: writes the buckets of the one account this fold was handed. await client.$executeRaw` INSERT INTO "mood_entry_rollups" ("user_id", "granularity", "bucket_start", "count", "mean", "min_score", "max_score", "sd", "computed_at") diff --git a/src/lib/workouts/climb-backfill.ts b/src/lib/workouts/climb-backfill.ts index c66983c84..d1c7e3ebb 100644 --- a/src/lib/workouts/climb-backfill.ts +++ b/src/lib/workouts/climb-backfill.ts @@ -18,6 +18,7 @@ * * Not an AI path: the number lands on the workout row, nothing else. */ +// account-deletion(file): a one-time column fill behind a cursor; it derives elevation from the route already stored and adds nothing the record did not hold, and a row skipped while its account is pending would stay empty after an undo. import type { PrismaClient } from "@/generated/prisma/client"; import { readRouteGeometry } from "@/lib/workouts/route-geometry-cipher"; import { routeClimbM } from "@/lib/workouts/route-climb"; diff --git a/tests/integration/account-deletion-grace.test.ts b/tests/integration/account-deletion-grace.test.ts index c7f854bc5..f1cc5bb29 100644 --- a/tests/integration/account-deletion-grace.test.ts +++ b/tests/integration/account-deletion-grace.test.ts @@ -285,6 +285,37 @@ describe("account deletion with a grace window", () => { const candidates = await findPregenerateCandidates(prisma, new Date(), 50); expect(candidates.map((c) => c.id)).toEqual([active.id]); + // The nightly score jobs: a cross-account `groupBy` the guard used to + // read as scoped because its `orderBy` names `userId`. + for (const userId of [pending.id, active.id]) { + await prisma.measurement.create({ + data: { + userId, + type: "HEART_RATE_VARIABILITY", + value: 48, + unit: "ms", + measuredAt: new Date(Date.now() - 3_600_000), + source: "MANUAL", + }, + }); + } + const { findStressScoreCandidates } = + await import("@/lib/jobs/stress-score"); + expect(await findStressScoreCandidates(prisma, new Date(), 50)).toEqual([ + active.id, + ]); + + // A job queued for it before the deletion completes without running. + const { dropJobsOfPendingAccounts } = + await import("@/lib/jobs/pending-deletion-gate"); + const queued = [ + { id: "j1", data: { userId: pending.id } }, + { id: "j2", data: { userId: active.id } }, + ]; + expect( + (await dropJobsOfPendingAccounts(queued)).live.map((j) => j.id), + ).toEqual(["j2"]); + // Nothing is delivered about or to it, whoever enumerated it. const { resolveNotificationDeliveryIdentity } = await import("@/lib/notifications/delivery-identity");