diff --git a/src/lib/__tests__/db.test.ts b/src/lib/__tests__/db.test.ts index d433f534d..ec872c51f 100644 --- a/src/lib/__tests__/db.test.ts +++ b/src/lib/__tests__/db.test.ts @@ -16,7 +16,7 @@ */ import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { getPoolMax } from "../db"; +import { buildSessionOptions, getPoolMax, getStatementTimeoutMs } from "../db"; describe("getPoolMax", () => { const originalEnv = process.env.DATABASE_POOL_MAX; @@ -62,3 +62,66 @@ describe("getPoolMax", () => { expect(getPoolMax()).toBeGreaterThanOrEqual(20); }); }); + +// A-1 — a per-session statement timeout so a single stuck query can't hold a +// pool slot forever and starve every DB-backed route. +describe("getStatementTimeoutMs", () => { + const originalEnv = process.env.DATABASE_STATEMENT_TIMEOUT_MS; + + beforeEach(() => { + delete process.env.DATABASE_STATEMENT_TIMEOUT_MS; + }); + + afterEach(() => { + if (originalEnv === undefined) { + delete process.env.DATABASE_STATEMENT_TIMEOUT_MS; + } else { + process.env.DATABASE_STATEMENT_TIMEOUT_MS = originalEnv; + } + }); + + it("defaults to 15000ms when the env var is unset", () => { + expect(getStatementTimeoutMs()).toBe(15_000); + }); + + it("honours a positive override", () => { + process.env.DATABASE_STATEMENT_TIMEOUT_MS = "30000"; + expect(getStatementTimeoutMs()).toBe(30_000); + }); + + it("allows 0 to disable the timeout (legacy unbounded behaviour)", () => { + process.env.DATABASE_STATEMENT_TIMEOUT_MS = "0"; + expect(getStatementTimeoutMs()).toBe(0); + }); + + it("falls back to the default on a malformed or negative value", () => { + process.env.DATABASE_STATEMENT_TIMEOUT_MS = "not-a-number"; + expect(getStatementTimeoutMs()).toBe(15_000); + process.env.DATABASE_STATEMENT_TIMEOUT_MS = "-5"; + expect(getStatementTimeoutMs()).toBe(15_000); + }); +}); + +describe("buildSessionOptions", () => { + const originalEnv = process.env.DATABASE_STATEMENT_TIMEOUT_MS; + + afterEach(() => { + if (originalEnv === undefined) { + delete process.env.DATABASE_STATEMENT_TIMEOUT_MS; + } else { + process.env.DATABASE_STATEMENT_TIMEOUT_MS = originalEnv; + } + }); + + it("emits both session timeouts as libpq startup options by default", () => { + delete process.env.DATABASE_STATEMENT_TIMEOUT_MS; + const opts = buildSessionOptions(); + expect(opts).toContain("-c statement_timeout=15000"); + expect(opts).toContain("-c idle_in_transaction_session_timeout=15000"); + }); + + it("returns undefined (no options) when the timeout is disabled", () => { + process.env.DATABASE_STATEMENT_TIMEOUT_MS = "0"; + expect(buildSessionOptions()).toBeUndefined(); + }); +}); diff --git a/src/lib/cycle/__tests__/day-log-write.test.ts b/src/lib/cycle/__tests__/day-log-write.test.ts new file mode 100644 index 000000000..8fdd8119e --- /dev/null +++ b/src/lib/cycle/__tests__/day-log-write.test.ts @@ -0,0 +1,156 @@ +/** + * Cycle day-log upsert helper — C-1 encryption write-path data-loss + * regression coverage. + * + * The partial-update upsert loads the existing row before writing. When the + * stored encrypted note / sensitive envelope cannot be decrypted (rotation + * gap, GCM corruption), the helper must NOT collapse the column to a re-encrypt + * of a failed soft-decrypt (which would permanently wipe the user's data when + * they edit an unrelated field). The stored ciphertext is preserved verbatim + * unless the user explicitly sets that field. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const db = vi.hoisted(() => ({ + cycleDayLog: { + findUnique: vi.fn(), + upsert: vi.fn(), + update: vi.fn(), + }, + cycleSymptom: { findMany: vi.fn() }, + cycleSymptomLink: { deleteMany: vi.fn(), createMany: vi.fn() }, +})); + +vi.mock("@/lib/db", () => ({ prisma: db })); +vi.mock("@/lib/cycle/profile", () => ({ + getOrCreateCycleProfile: vi.fn(async () => ({ + sensitiveCategoryEncryption: true, + })), +})); + +// `decrypt` round-trips an `enc:` prefix and THROWS on any other ciphertext, +// simulating an undecryptable column. The write path must never let that throw +// reach the column write — it preserves the stored ciphertext verbatim. +vi.mock("@/lib/crypto", () => ({ + encrypt: (s: string) => `enc:${s}`, + decrypt: (c: string) => { + if (!c.startsWith("enc:")) throw new Error("undecryptable"); + return c.replace(/^enc:/, ""); + }, +})); + +import { upsertCycleDayLog } from "@/lib/cycle/day-log-write"; +import type { CycleDayLogInput } from "@/lib/validations/cycle"; + +function existingRow(overrides: Record = {}) { + return { + id: "cdl-1", + date: "2026-06-16", + flow: null, + intermenstrualBleeding: false, + basalBodyTempC: null, + temperatureExcluded: false, + ovulationTest: null, + cervicalMucus: null, + cervixPosition: null, + cervixFirmness: null, + cervixOpening: null, + sexualActivity: false, + protectedSex: null, + pregnancyTest: null, + progesteroneTest: null, + contraceptive: null, + sensitiveEncrypted: null, + notesEncrypted: null, + deletedAt: null, + ...overrides, + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + db.cycleDayLog.upsert.mockResolvedValue({ id: "cdl-1" }); +}); + +describe("upsertCycleDayLog — C-1 undecryptable note preservation", () => { + it("preserves an undecryptable note verbatim when an unrelated field is edited", async () => { + db.cycleDayLog.findUnique.mockResolvedValue( + existingRow({ notesEncrypted: "CORRUPT-NOTE-CIPHERTEXT" }), + ); + // Edit only `flow`; `note` is omitted. + const entry = { + date: "2026-06-16", + source: "MANUAL", + flow: "LIGHT", + } as unknown as CycleDayLogInput; + + await upsertCycleDayLog("u1", entry, "Europe/Berlin"); + + const args = db.cycleDayLog.upsert.mock.calls[0][0]; + // The stored ciphertext is carried through unchanged — not nulled, not + // re-encrypted from a failed soft-decrypt. + expect(args.update.notesEncrypted).toBe("CORRUPT-NOTE-CIPHERTEXT"); + expect(args.update.flow).toBe("LIGHT"); + }); + + it("clears the note on an explicit null even when the stored one is undecryptable", async () => { + db.cycleDayLog.findUnique.mockResolvedValue( + existingRow({ notesEncrypted: "CORRUPT-NOTE-CIPHERTEXT" }), + ); + const entry = { + date: "2026-06-16", + source: "MANUAL", + note: null, + } as unknown as CycleDayLogInput; + + await upsertCycleDayLog("u1", entry, "Europe/Berlin"); + + const args = db.cycleDayLog.upsert.mock.calls[0][0]; + expect(args.update.notesEncrypted).toBeNull(); + }); +}); + +describe("upsertCycleDayLog — C-1 undecryptable sensitive envelope preservation", () => { + it("preserves an undecryptable sensitive envelope verbatim when an unrelated field is edited", async () => { + db.cycleDayLog.findUnique.mockResolvedValue( + existingRow({ sensitiveEncrypted: "CORRUPT-ENVELOPE-CIPHERTEXT" }), + ); + // Edit only a non-sensitive field; no sensitive field supplied. + const entry = { + date: "2026-06-16", + source: "MANUAL", + basalBodyTempC: 36.5, + } as unknown as CycleDayLogInput; + + await upsertCycleDayLog("u1", entry, "Europe/Berlin"); + + const args = db.cycleDayLog.upsert.mock.calls[0][0]; + // The envelope ciphertext is preserved, not overwritten with a defaults + // envelope minted from a failed soft-decrypt. + expect(args.update.sensitiveEncrypted).toBe("CORRUPT-ENVELOPE-CIPHERTEXT"); + expect(args.update.basalBodyTempC).toBe(36.5); + }); + + it("re-encrypts the envelope when a sensitive field is explicitly supplied", async () => { + db.cycleDayLog.findUnique.mockResolvedValue( + existingRow({ sensitiveEncrypted: "CORRUPT-ENVELOPE-CIPHERTEXT" }), + ); + const entry = { + date: "2026-06-16", + source: "MANUAL", + sexualActivity: true, + } as unknown as CycleDayLogInput; + + await upsertCycleDayLog("u1", entry, "Europe/Berlin"); + + const args = db.cycleDayLog.upsert.mock.calls[0][0]; + // An explicit sensitive write mints a fresh envelope carrying the user's + // intent — the stored (undecryptable) one is replaced on purpose. + expect(args.update.sensitiveEncrypted).not.toBe( + "CORRUPT-ENVELOPE-CIPHERTEXT", + ); + expect(String(args.update.sensitiveEncrypted)).toContain( + '"sexualActivity":true', + ); + }); +}); diff --git a/src/lib/cycle/day-log-write.ts b/src/lib/cycle/day-log-write.ts index 1113c506f..10f7bc324 100644 --- a/src/lib/cycle/day-log-write.ts +++ b/src/lib/cycle/day-log-write.ts @@ -114,13 +114,22 @@ export async function replaceSymptomLinks( } } -/** Decrypt the sensitive envelope fail-soft (null on missing / undecryptable). */ -function decryptSensitive(envelope: string | null): Partial { - if (!envelope) return {}; +/** + * Decrypt the sensitive envelope fail-soft. Returns `{ ok: true, values }` on a + * clean decrypt, `{ ok: false }` when an envelope is present but undecryptable + * (rotation gap, GCM corruption). The caller MUST treat `ok: false` as + * "do not re-write this column" — never as "the field is empty". + */ +function decryptSensitive( + envelope: string, +): { ok: true; values: Partial } | { ok: false } { try { - return JSON.parse(decrypt(envelope)) as Partial; + return { + ok: true, + values: JSON.parse(decrypt(envelope)) as Partial, + }; } catch { - return {}; + return { ok: false }; } } @@ -259,21 +268,31 @@ async function writeDayLog(args: WriteArgs): Promise { // Decrypt the existing note (fail-soft) for the change/duplicate split — // re-encrypting the same plaintext yields a different ciphertext (random - // GCM IV), so a ciphertext compare would always read "changed". + // GCM IV), so a ciphertext compare would always read "changed". This is a + // DISPLAY-ONLY read (a `null` here only affects the change flag, never the + // written column — see the note-write below), so soft-decrypt is safe. const existingNote = decryptNoteSoft(existing?.notesEncrypted ?? null); // The resolved plaintext sensitive values for THIS write, after partial // merge against what's already stored (decrypted from the envelope when - // the stored row was encrypted). - const storedSensitive: SensitiveFields = existing + // the stored row was encrypted). `storedDecryptable` is false when the + // existing envelope could not be decrypted — in that case the merged + // values are DEFAULTS, not the real stored data, so they must never be + // re-encrypted back over the column (that would wipe the envelope). + const merged = existing ? mergeStoredSensitive(existing) : { - sexualActivity: false, - protectedSex: null, - pregnancyTest: null, - progesteroneTest: null, - contraceptive: null, + values: { + sexualActivity: false, + protectedSex: null, + pregnancyTest: null, + progesteroneTest: null, + contraceptive: null, + }, + decryptable: true, }; + const storedSensitive: SensitiveFields = merged.values; + const storedSensitiveDecryptable = merged.decryptable; const resolvedSensitive: SensitiveFields = { sexualActivity: @@ -298,13 +317,29 @@ async function writeDayLog(args: WriteArgs): Promise { : storedSensitive.contraceptive, }; - // Note resolution with partial merge: an omitted `note` keeps the stored - // value; an explicit null clears it. - const incomingNote = + // Note resolution with partial merge. When `note` is supplied, encrypt it + // (an explicit empty/null clears the column). When `note` is OMITTED, the + // existing ciphertext is preserved VERBATIM — never decrypt→re-encrypt, and + // never collapse an undecryptable existing value to null. A soft-decrypt that + // returns null on an undecryptable column (rotation gap, GCM corruption) must + // not be allowed to permanently wipe the stored note when an unrelated field + // is the only thing being edited (mirrors the v1.18.1 53-column rotation fix: + // a write path never re-writes an `*Encrypted` column from a soft-decrypt + // result). + const notesEncrypted: string | null = entry.note !== undefined - ? (entry.note ?? null) - : decryptNoteSoft(existing?.notesEncrypted ?? null); - const notesEncrypted = incomingNote ? encrypt(incomingNote) : null; + ? entry.note + ? encrypt(entry.note) + : null + : (existing?.notesEncrypted ?? null); + + // Note change detection for the `changed`/`duplicate` split. An omitted + // `note` preserves the column verbatim, so it can never count as a change. + // A supplied `note` is compared against the soft-decrypted existing value + // (display-only) — a ciphertext compare would always read "changed" (random + // GCM IV). + const incomingNote = entry.note !== undefined ? (entry.note ?? null) : null; + const noteChanged = entry.note !== undefined && incomingNote !== existingNote; // Non-sensitive fields, partial-merged against the stored row. const flow = @@ -342,11 +377,35 @@ async function writeDayLog(args: WriteArgs): Promise { ? (entry.cervixOpening ?? null) : (existing?.cervixOpening ?? null); + // Was any of the five sensitive fields actually supplied in THIS update? + // Used to decide whether an undecryptable existing envelope may be preserved + // verbatim (an unrelated-field edit) or must be rewritten (the user is + // explicitly setting sensitive data). + const sensitiveFieldSupplied = + entry.sexualActivity !== undefined || + entry.protectedSex !== undefined || + entry.pregnancyTest !== undefined || + entry.progesteroneTest !== undefined || + entry.contraceptive !== undefined; + // Split the sensitive fields between plaintext columns and the envelope // depending on the flag. When encrypting, the plaintext columns are NULL // (drop out of rollup/correlation) and the envelope carries the JSON. + // + // Data-loss guard: when the stored envelope could NOT be decrypted, the + // merged `resolvedSensitive` carries DEFAULTS, not the real stored values. + // Re-encrypting them over the column would permanently wipe the user's + // sensitive data on an edit of an unrelated field. So when encrypting, the + // existing envelope is undecryptable, and no sensitive field is supplied in + // this update, preserve the existing ciphertext VERBATIM (mirrors the + // v1.18.1 53-column rotation fix). An explicit sensitive-field write still + // re-encrypts the user's intent. + const preserveExistingEnvelope = + encryptSensitive && !storedSensitiveDecryptable && !sensitiveFieldSupplied; const sensitiveEncrypted = encryptSensitive - ? encrypt(JSON.stringify(resolvedSensitive)) + ? preserveExistingEnvelope + ? (existing?.sensitiveEncrypted ?? null) + : encrypt(JSON.stringify(resolvedSensitive)) : null; const sensitivePlaintext = encryptSensitive ? { @@ -391,12 +450,18 @@ async function writeDayLog(args: WriteArgs): Promise { (existing.cervixPosition ?? null) !== cervixPosition || (existing.cervixFirmness ?? null) !== cervixFirmness || (existing.cervixOpening ?? null) !== cervixOpening || - storedSensitive.sexualActivity !== resolvedSensitive.sexualActivity || - storedSensitive.protectedSex !== resolvedSensitive.protectedSex || - storedSensitive.pregnancyTest !== resolvedSensitive.pregnancyTest || - storedSensitive.progesteroneTest !== resolvedSensitive.progesteroneTest || - storedSensitive.contraceptive !== resolvedSensitive.contraceptive || - existingNote !== incomingNote || + // When the stored envelope was undecryptable and we preserve it + // verbatim (no sensitive field supplied), the sensitive values cannot + // be compared meaningfully — skip them. An explicit sensitive write + // compares the resolved values against the (default) stored ones. + (!preserveExistingEnvelope && + (storedSensitive.sexualActivity !== resolvedSensitive.sexualActivity || + storedSensitive.protectedSex !== resolvedSensitive.protectedSex || + storedSensitive.pregnancyTest !== resolvedSensitive.pregnancyTest || + storedSensitive.progesteroneTest !== + resolvedSensitive.progesteroneTest || + storedSensitive.contraceptive !== resolvedSensitive.contraceptive)) || + noteChanged || existing.deletedAt !== null || (Boolean(entry.externalId) && existing.date !== entry.date); } @@ -492,23 +557,49 @@ function isUniqueViolation(err: unknown): boolean { ); } -/** Resolve the stored plaintext sensitive fields (envelope OR columns). */ -function mergeStoredSensitive(existing: ExistingRow): SensitiveFields { +/** + * Resolve the stored plaintext sensitive fields (envelope OR columns). + * `decryptable` is false ONLY when an envelope is present but could not be + * decrypted — the returned `values` are then DEFAULTS and the caller must not + * re-encrypt them back over the column. + */ +function mergeStoredSensitive(existing: ExistingRow): { + values: SensitiveFields; + decryptable: boolean; +} { if (existing.sensitiveEncrypted) { const dec = decryptSensitive(existing.sensitiveEncrypted); + if (!dec.ok) { + return { + values: { + sexualActivity: false, + protectedSex: null, + pregnancyTest: null, + progesteroneTest: null, + contraceptive: null, + }, + decryptable: false, + }; + } return { - sexualActivity: dec.sexualActivity ?? false, - protectedSex: dec.protectedSex ?? null, - pregnancyTest: dec.pregnancyTest ?? null, - progesteroneTest: dec.progesteroneTest ?? null, - contraceptive: dec.contraceptive ?? null, + values: { + sexualActivity: dec.values.sexualActivity ?? false, + protectedSex: dec.values.protectedSex ?? null, + pregnancyTest: dec.values.pregnancyTest ?? null, + progesteroneTest: dec.values.progesteroneTest ?? null, + contraceptive: dec.values.contraceptive ?? null, + }, + decryptable: true, }; } return { - sexualActivity: existing.sexualActivity, - protectedSex: existing.protectedSex, - pregnancyTest: existing.pregnancyTest, - progesteroneTest: existing.progesteroneTest, - contraceptive: existing.contraceptive, + values: { + sexualActivity: existing.sexualActivity, + protectedSex: existing.protectedSex, + pregnancyTest: existing.pregnancyTest, + progesteroneTest: existing.progesteroneTest, + contraceptive: existing.contraceptive, + }, + decryptable: true, }; } diff --git a/src/lib/db.ts b/src/lib/db.ts index 7097d0a83..01ab1607f 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -13,14 +13,14 @@ const globalForPrisma = globalThis as unknown as { * (`.planning/round-v1439-empirical-trace.md` § B2) showed thick * `/api/analytics` holding ≥ 8 of the 10 default pool slots for * 6.5 s on the maintainer's 347k-row tenant, starving every other Wave-B and - * Wave-C useQuery fan-out for the duration. Coolify's Postgres - * container runs the stock `max_connections` 100 ceiling, so a - * 20-slot Node-side pool sits well under the server's hard cap with - * plenty of headroom for concurrent power-users (≤ 5 simultaneous - * tenants × 20 = 100). Complements the W-POOL `p-limit(4)` cap on - * the analytics fan-out itself — the bounded concurrency keeps any - * single analytics call to ≤ 4 slots, so the remaining 16 stay - * available for every other dashboard query. + * Wave-C useQuery fan-out for the duration. The production Postgres runs a + * `max_connections` of 200 (raised from the stock 100), so a 20-slot + * Node-side pool sits well under the server's hard cap with plenty of + * headroom for concurrent power-users. Note the web + worker containers + * share this one Postgres, so the effective ceiling covers both pools. + * Complements the W-POOL `p-limit(4)` cap on the analytics fan-out itself — + * the bounded concurrency keeps any single analytics call to ≤ 4 slots, so + * the remaining 16 stay available for every other dashboard query. * * Env-overridable via `DATABASE_POOL_MAX` so an operator running * with a smaller Postgres `max_connections` can dial down without a @@ -35,10 +35,51 @@ export function getPoolMax(): number { return 20; } +/** + * Per-session DB statement timeout in milliseconds. + * + * Without this a single pathological query or lock wait holds one of the + * (default 20) pool slots indefinitely; 20 such queries permanently exhaust + * the pool and every DB-backed route stops serving — the single highest-impact + * availability gap (A-1). `statement_timeout` caps how long the server runs one + * statement; `idle_in_transaction_session_timeout` reaps a connection wedged + * mid-transaction (a client that opened a tx and stalled). Both are applied at + * connection-establishment via the libpq `options` startup parameter, so every + * pooled session inherits them. + * + * Env-overridable via `DATABASE_STATEMENT_TIMEOUT_MS`. Set to `0` to disable + * (legacy unbounded behaviour). Default 15 s — generous for the rollup-backed + * read paths while still bounding a runaway query well before it can starve the + * pool. Heavy admin one-shots (drain/backfill) run on the worker, not the web + * pool, and can dial their own timeout if needed. + */ +export function getStatementTimeoutMs(): number { + const raw = process.env.DATABASE_STATEMENT_TIMEOUT_MS; + if (raw !== undefined && raw !== "") { + const parsed = Number.parseInt(raw, 10); + if (Number.isFinite(parsed) && parsed >= 0) return parsed; + } + return 15_000; +} + +/** + * Build the libpq `options` startup string applying the session timeouts, or + * `undefined` when disabled (timeout 0). Passed straight through `PrismaPg` to + * the underlying `pg.Pool`, which forwards it as the connection's `options` + * startup parameter so every session is timeout-bounded from the first query. + */ +export function buildSessionOptions(): string | undefined { + const timeoutMs = getStatementTimeoutMs(); + if (timeoutMs <= 0) return undefined; + return `-c statement_timeout=${timeoutMs} -c idle_in_transaction_session_timeout=${timeoutMs}`; +} + function createPrismaClient() { + const sessionOptions = buildSessionOptions(); const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL!, max: getPoolMax(), + ...(sessionOptions ? { options: sessionOptions } : {}), }); return new PrismaClient({ adapter }); } diff --git a/src/lib/illness/__tests__/day-log-write.test.ts b/src/lib/illness/__tests__/day-log-write.test.ts index f7eb13b7b..9d63d2f43 100644 --- a/src/lib/illness/__tests__/day-log-write.test.ts +++ b/src/lib/illness/__tests__/day-log-write.test.ts @@ -25,10 +25,17 @@ const db = vi.hoisted(() => { }); vi.mock("@/lib/db", () => ({ prisma: db })); +// `decryptFromBytes` throws on any ciphertext NOT minted by this mock's +// `encryptToBytes` — simulating an undecryptable column (rotation gap / GCM +// corruption). The write path must never call it (it preserves ciphertext +// verbatim), so a throw here that reaches the column write is a regression. vi.mock("@/lib/ai/coach/bytes-codec", () => ({ encryptToBytes: (s: string) => new Uint8Array(Buffer.from(`enc:${s}`)), - decryptFromBytes: (b: Uint8Array) => - Buffer.from(b).toString("utf8").replace(/^enc:/, ""), + decryptFromBytes: (b: Uint8Array) => { + const s = Buffer.from(b).toString("utf8"); + if (!s.startsWith("enc:")) throw new Error("undecryptable"); + return s.replace(/^enc:/, ""); + }, })); import { @@ -188,4 +195,29 @@ describe("upsertIllnessDayLog", () => { expect(args.create.noteEncrypted).toBeInstanceOf(Uint8Array); expect(db.illnessSymptomLink.deleteMany).not.toHaveBeenCalled(); }); + + // C-1 data-loss regression: editing an unrelated field on a row whose + // encrypted note CANNOT be decrypted must preserve the existing ciphertext + // byte-for-byte — never null it from a failed soft-decrypt. + it("preserves an undecryptable note byte-for-byte when an unrelated field is edited", async () => { + const corrupt = new Uint8Array(Buffer.from("CORRUPT-CIPHERTEXT")); + db.illnessDayLog.findUnique.mockResolvedValue({ + id: "dl-1", + functionalImpact: 1, + feverC: null, + noteEncrypted: corrupt, + }); + // Only fever is supplied; `note` is omitted. + await upsertIllnessDayLog( + "u1", + "ep1", + { date: "2026-06-16", feverC: 38 }, + null, + ); + const args = db.illnessDayLog.upsert.mock.calls[0][0]; + // The stored ciphertext is carried through verbatim, never decrypted+ + // re-encrypted and never nulled. + expect(args.update.noteEncrypted).toBe(corrupt); + expect(args.update.feverC).toBe(38); + }); }); diff --git a/src/lib/illness/day-log-write.ts b/src/lib/illness/day-log-write.ts index a24311c07..6a1f01526 100644 --- a/src/lib/illness/day-log-write.ts +++ b/src/lib/illness/day-log-write.ts @@ -15,7 +15,7 @@ * dropped silently. */ import { prisma } from "@/lib/db"; -import { decryptFromBytes, encryptToBytes } from "@/lib/ai/coach/bytes-codec"; +import { encryptToBytes } from "@/lib/ai/coach/bytes-codec"; import type { IllnessDayLogInput } from "@/lib/validations/illness"; import { dayLogSymptomInclude, toIllnessDayLogDTO } from "@/lib/illness/dto"; import type { IllnessDayLogDTO } from "@/lib/illness/dto"; @@ -88,16 +88,6 @@ export async function replaceIllnessSymptomLinks( } } -/** Decrypt a Bytes note fail-soft (null on missing / undecryptable). */ -function decryptNoteSoft(noteEncrypted: Uint8Array | null): string | null { - if (!noteEncrypted || noteEncrypted.byteLength === 0) return null; - try { - return decryptFromBytes(noteEncrypted); - } catch { - return null; - } -} - /** * Upsert one illness day-log on the `(episodeId, date)` key. The caller is * responsible for verifying the episode is owned + live before calling. @@ -136,11 +126,21 @@ export async function upsertIllnessDayLog( ? (entry.feverC ?? null) : (existing?.feverC ?? null); - const incomingNote = + // Note resolution with partial merge. When `note` is supplied, encrypt it + // (an explicit empty/null clears the column). When `note` is OMITTED, the + // existing ciphertext is preserved BYTE-FOR-BYTE — never decrypt→re-encrypt, + // and never collapse an undecryptable existing value to null. A soft-decrypt + // that returns null on an undecryptable column (rotation gap, GCM + // corruption) must not be allowed to permanently wipe the stored note when + // an unrelated field is the only thing being edited (mirrors the v1.18.1 + // 53-column rotation fix: a write path never re-writes an `*Encrypted` + // column from a soft-decrypt result). + const noteEncrypted = entry.note !== undefined - ? (entry.note ?? null) - : decryptNoteSoft(existing?.noteEncrypted ?? null); - const noteEncrypted = incomingNote ? encryptToBytes(incomingNote) : null; + ? entry.note + ? encryptToBytes(entry.note) + : null + : (existing?.noteEncrypted ?? null); const upserted = await tx.illnessDayLog.upsert({ where, diff --git a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts index 0efaf57a9..d391105e5 100644 --- a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts +++ b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts @@ -342,16 +342,18 @@ describe("drainPerSampleCumulative — cutoffHours", () => { // A1 — a late watch sync after a day's total was already collapsed must // FOLD the new per-sample rows into the existing total, never overwrite -// it with just the partial late sum (the original samples are gone). +// it with just the partial late sum (the original samples are gone). The +// adopt-in-place write resolves the canonical row by the index-A `stats:` +// externalId first, then adopts it in place (`update`) or mints a fresh row +// (`create`). describe("drainPerSampleCumulative — late-sync fold into existing total", () => { - function buildFoldMock(existingValue: number | null) { + function buildFoldMock(existingTotal: number | null) { const findManyUser = vi .fn() .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]); // Only ACTIVITY_STEPS yields the late per-sample rows; every other - // cumulative type returns an empty scan so the run stays a single - // bucket. + // cumulative type returns an empty scan so the run stays a single bucket. const findManyMeasurement = vi.fn( async (args: { where: { type: string } }) => { if (args.where.type === "ACTIVITY_STEPS") { @@ -376,18 +378,32 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = }, ); - const upsert = vi.fn().mockResolvedValue({}); + const update = vi.fn().mockResolvedValue({ id: "stats-row" }); + const create = vi.fn().mockResolvedValue({ id: "stats-row-new" }); const deleteMany = vi.fn().mockResolvedValue({ count: 2 }); - const findUnique = vi - .fn() - .mockResolvedValue( - existingValue === null ? null : { value: existingValue }, - ); - const findFirst = vi.fn().mockResolvedValue({ unit: "count" }); - const tx = { - measurement: { upsert, deleteMany, findUnique, findFirst }, - }; + // `findFirst` serves three roles inside `writeDay`: + // 1. index-A lookup (filters on `externalId`) → the existing collapsed + // total when one exists. + // 2. index-B slot lookup (filters on `measuredAt`) → null here (no row + // sits on the canonical instant). + // 3. `resolveCanonicalUnit` (filters on neither, selects `unit`) → unit. + const findFirst = vi.fn( + async (args: { + where: Record; + select?: Record; + }) => { + if ("externalId" in args.where) { + return existingTotal === null + ? null + : { id: "stats-row", value: existingTotal }; + } + if ("measuredAt" in args.where) return null; // no index-B collision + return { unit: "count" }; // resolveCanonicalUnit + }, + ); + + const tx = { measurement: { update, create, deleteMany, findFirst } }; return { prisma: { @@ -397,39 +413,173 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = cb(tx), ), } as unknown as PrismaClient, - upsert, - findUnique, + update, + create, }; } it("folds late samples into a pre-existing collapsed total (no shrink)", async () => { // Day already collapsed to 9000 on an earlier run; a late sync of // 300 + 200 = 500 must take the row to 9500, not down to 500. - const { prisma, upsert, findUnique } = buildFoldMock(9000); + const { prisma, update, create } = buildFoldMock(9000); await drainPerSampleCumulative(prisma, { log: () => {} }); - expect(findUnique).toHaveBeenCalledTimes(1); - expect(upsert).toHaveBeenCalledTimes(1); - const upsertArg = upsert.mock.calls[0]?.[0] as { - update: { value: number }; - }; - expect(upsertArg.update.value).toBe(9500); + // The index-A row is adopted in place — updated, never re-created. + expect(create).not.toHaveBeenCalled(); + expect(update).toHaveBeenCalledTimes(1); + const updateArg = update.mock.calls[0]?.[0] as { data: { value: number } }; + expect(updateArg.data.value).toBe(9500); }); it("mints the partial sum when no collapsed total exists yet", async () => { - const { prisma, upsert } = buildFoldMock(null); + const { prisma, update, create } = buildFoldMock(null); await drainPerSampleCumulative(prisma, { log: () => {} }); - const upsertArg = upsert.mock.calls[0]?.[0] as { - create: { value: number }; - update: { value: number }; + // Fresh day: no canonical row to adopt → a new row is created with the + // raw sum. + expect(update).not.toHaveBeenCalled(); + expect(create).toHaveBeenCalledTimes(1); + const createArg = create.mock.calls[0]?.[0] as { data: { value: number } }; + expect(createArg.data.value).toBe(500); + }); +}); + +// Root cause of the v1.18.10 bloat: a single day whose mint collided with the +// canonical-noon unique index threw and aborted the ENTIRE global walk (3 s +// die-early, 0 collapsed). The per-day error boundary must step the poisoned +// day over so every other day still collapses; the adopt-in-place path must +// absorb an index-B slot collision rather than throw at all. +describe("drainPerSampleCumulative — colliding day does not abort the walk", () => { + it("steps over a poisoned day and still collapses the others", async () => { + const findManyUser = vi + .fn() + .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]); + + // Two ACTIVITY_STEPS days: 2026-05-16 (poison) and 2026-05-17 (clean). + const findManyMeasurement = vi.fn( + async (args: { where: { type: string } }) => { + if (args.where.type === "ACTIVITY_STEPS") { + return [ + { + id: "d16-a", + type: "ACTIVITY_STEPS", + value: 1000, + measuredAt: new Date("2026-05-16T09:00:00.000Z"), + externalId: "hk-uuid-16a", + }, + { + id: "d17-a", + type: "ACTIVITY_STEPS", + value: 2000, + measuredAt: new Date("2026-05-17T09:00:00.000Z"), + externalId: "hk-uuid-17a", + }, + ]; + } + return []; + }, + ); + + const create = vi.fn(async (args: { data: { measuredAt: Date } }) => { + // The 2026-05-16 day mint throws a NON-P2002 error on every attempt — + // a genuinely poisoned day the adopt path can't recover. Berlin local + // noon on 2026-05-16 (CEST) is 10:00 UTC. + const iso = args.data.measuredAt.toISOString(); + if (iso === "2026-05-16T10:00:00.000Z") { + throw new Error("simulated poison-day write failure"); + } + return { id: "stats-row-new" }; + }); + const update = vi.fn().mockResolvedValue({ id: "stats-row" }); + const deleteMany = vi.fn().mockResolvedValue({ count: 1 }); + const findFirst = vi.fn( + async (args: { where: Record }) => { + if ("externalId" in args.where) return null; // no existing total + if ("measuredAt" in args.where) return null; // no index-B collision + return { unit: "count" }; + }, + ); + + const tx = { measurement: { update, create, deleteMany, findFirst } }; + const prisma = { + user: { findMany: findManyUser }, + measurement: { findMany: findManyMeasurement }, + $transaction: vi.fn(async (cb: (t: typeof tx) => Promise) => + cb(tx), + ), + } as unknown as PrismaClient; + + const summary = await drainPerSampleCumulative(prisma, { log: () => {} }); + + // The poisoned day is stepped over, NOT rethrown; the clean day collapses. + expect(summary.totals.daysFailed).toBe(1); + expect(summary.totals.bucketsCollapsed).toBe(1); + expect(summary.totals.usersScanned).toBe(1); + // The clean 2026-05-17 mint went through. + expect(create).toHaveBeenCalledTimes(2); // both days attempt a create + }); + + it("adopts an index-B canonical-noon row in place instead of colliding", async () => { + const findManyUser = vi + .fn() + .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]); + + const findManyMeasurement = vi.fn( + async (args: { where: { type: string } }) => { + if (args.where.type === "ACTIVITY_STEPS") { + return [ + { + id: "samp-1", + type: "ACTIVITY_STEPS", + value: 1500, + measuredAt: new Date("2026-05-16T09:00:00.000Z"), + externalId: "hk-uuid-1", + }, + ]; + } + return []; + }, + ); + + const update = vi.fn().mockResolvedValue({ id: "slot-row" }); + const create = vi.fn().mockResolvedValue({ id: "should-not-be-called" }); + const deleteMany = vi.fn().mockResolvedValue({ count: 1 }); + // No index-A stats row, but a DIFFERENT row already sits on the canonical + // local-noon instant (index B) — e.g. a manual daily entry. The drain must + // adopt it in place, never `create` (which would P2002 against index B). + const findFirst = vi.fn( + async (args: { where: Record }) => { + if ("externalId" in args.where) return null; + if ("measuredAt" in args.where) return { id: "slot-row" }; + return { unit: "count" }; + }, + ); + + const tx = { measurement: { update, create, deleteMany, findFirst } }; + const prisma = { + user: { findMany: findManyUser }, + measurement: { findMany: findManyMeasurement }, + $transaction: vi.fn(async (cb: (t: typeof tx) => Promise) => + cb(tx), + ), + } as unknown as PrismaClient; + + const summary = await drainPerSampleCumulative(prisma, { log: () => {} }); + + expect(create).not.toHaveBeenCalled(); + expect(update).toHaveBeenCalledTimes(1); + const updateArg = update.mock.calls[0]?.[0] as { + where: { id: string }; + data: { externalId: string; value: number }; }; - // Fresh day: create branch carries the raw sum; update branch (taken - // on an idempotent re-run with no prior total) also resolves to the - // raw sum since `existing` was null. - expect(upsertArg.create.value).toBe(500); - expect(upsertArg.update.value).toBe(500); + // The slot row is adopted: stamped with the target stats externalId and + // the bucket's value, no collision, no throw. + expect(updateArg.where.id).toBe("slot-row"); + expect(updateArg.data.value).toBe(1500); + expect(updateArg.data.externalId).toContain("stats:"); + expect(summary.totals.daysFailed).toBe(0); + expect(summary.totals.bucketsCollapsed).toBe(1); }); }); diff --git a/src/lib/measurements/drain-per-sample-cumulative.ts b/src/lib/measurements/drain-per-sample-cumulative.ts index 8532dc035..9bd92ca3b 100644 --- a/src/lib/measurements/drain-per-sample-cumulative.ts +++ b/src/lib/measurements/drain-per-sample-cumulative.ts @@ -33,6 +33,7 @@ */ import { Prisma } from "@/generated/prisma/client"; import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import { isP2002 as isUniqueConstraintViolation } from "@/lib/prisma-errors"; import { CUMULATIVE_HK_TYPES, @@ -113,6 +114,12 @@ export interface DrainSummary { bucketsCollapsed: number; perSampleRowsDeleted: number; dailyRowsUpserted: number; + /** + * Day buckets whose write threw a non-recoverable error and were stepped + * over by the per-day boundary. A single poisoned day no longer aborts the + * whole global walk — every other day still collapses. + */ + daysFailed: number; }; } @@ -186,6 +193,7 @@ export async function drainPerSampleCumulative( bucketsCollapsed: 0, perSampleRowsDeleted: 0, dailyRowsUpserted: 0, + daysFailed: 0, }, }; @@ -226,85 +234,127 @@ export async function drainPerSampleCumulative( dayRows, sourceRowIds, }): Promise => { - let removed = 0; - await pc.$transaction(async (tx) => { - // Fold late per-sample rows INTO an existing collapsed daily - // total rather than overwriting it. `bucketRowsByDay` strips - // already-collapsed `stats:` rows from the bucket, so when a day - // was collapsed on an earlier run and fresh per-sample rows - // arrive late (a trailing watch sync after the day's total was - // already minted), this bucket holds ONLY those late samples. - // A blind `update: { value: reducedValue }` would replace the - // full day total with just the partial late sum — and the - // original contributing samples were already hard-deleted, so - // the lost count is unrecoverable. - // - // The probe runs inside the same transaction as the upsert + - // delete so the read-modify-write of the total is atomic against - // a concurrent drain. The late samples are genuine additional - // readings (unlike the legacy-step pass, where the existing - // total is HealthKit's own aggregate and the legacy rows would - // double-count — so that sibling deliberately does NOT fold), - // therefore the correct merge is `existing + late sum`. The - // per-sample rows are hard-deleted either way so a re-run cannot - // re-accumulate them. - const existing = await tx.measurement.findUnique({ - where: { - userId_type_source_externalId: { + // Adopt-in-place canonical-noon resolution + a single P2002 retry, + // mirroring `dense-intraday-retention.ts`. Two unique indexes can + // collide when minting the daily-stats row: + // A) `(userId, type, source, externalId)` — a row already carrying + // the target `stats:` externalId (a prior collapse of this day). + // B) `(userId, type, measuredAt, source, sleepStage)` NULLS NOT + // DISTINCT — ANY row sitting on the canonical local-noon instant, + // which may carry a DIFFERENT externalId (a per-sample row that + // happened at local noon and is in this very bucket, or a manual + // daily row). A blind `create` against index B aborts the whole + // transaction, and — before the per-day error boundary below — the + // first colliding day aborted the ENTIRE global walk (3 s die-early, + // 0 collapsed). Resolve the canonical row by index A first (it IS + // the daily row by construction), else the index-B slot row, and + // adopt it in place instead of colliding. + const foldOnce = async (): Promise => { + let removed = 0; + await pc.$transaction(async (tx) => { + // Index-A row: already carries the target `stats:` externalId. When + // present, this is the existing collapsed daily total. + const eidRow = await tx.measurement.findFirst({ + where: { userId, type, source: "APPLE_HEALTH", externalId }, + select: { id: true, value: true }, + orderBy: { id: "asc" }, + }); + // Index-B row: occupies the canonical local-noon instant + // (`sleepStage` is NULL for these cumulative types). At most one. + const slotRow = await tx.measurement.findFirst({ + where: { userId, type, source: "APPLE_HEALTH", - externalId, + measuredAt: canonicalTimestamp, + sleepStage: null, }, - }, - select: { value: true }, - }); - const mergedValue = - existing !== null ? existing.value + reducedValue : reducedValue; + select: { id: true }, + orderBy: { id: "asc" }, + }); - // Upsert the daily-aggregated row first, then drop the - // per-sample rows. The unique index - // (userId, type, source, externalId) makes the upsert - // idempotent across re-runs. - await tx.measurement.upsert({ - where: { - userId_type_source_externalId: { - userId, - type, - source: "APPLE_HEALTH", - externalId, - }, - }, - create: { - userId, - type, - value: reducedValue, - // pick the canonical unit from an existing row; the - // per-sample rows all carry the same unit on a given type. - // dayRows always has ≥1 row (empty buckets are skipped). - unit: - dayRows[0]?.value !== undefined - ? await resolveCanonicalUnit(tx, userId, type) - : "count", - source: "APPLE_HEALTH", - measuredAt: canonicalTimestamp, - externalId, - }, - update: { - value: mergedValue, - measuredAt: canonicalTimestamp, - }, - }); + // Prefer the externalId-carrying row (index A) — adopting a sibling + // and stamping the target externalId onto it would collide with it. + const adoptTarget = eidRow ?? slotRow; - // Delete the per-sample rows that contributed to the sum. - // Using `id IN (...)` is bounded by the per-bucket cap (the - // largest cumulative bucket in real data is a ~1 440-row - // stepCount day on a phone-only user). - const del = await tx.measurement.deleteMany({ - where: { id: { in: sourceRowIds } }, + // Fold semantics. When the index-A `stats:` total already exists, + // late per-sample rows are GENUINE additional readings, so the + // correct merge is `existing + late sum` (a blind overwrite with the + // partial late sum would shrink the day — the original samples are + // already hard-deleted). When adopting a NON-stats slot row (a + // per-sample row that fell on local noon, itself part of this + // bucket), its value is already inside `reducedValue`, so the merged + // value is just `reducedValue`. A fresh day mints `reducedValue`. + const mergedValue = + eidRow !== null ? eidRow.value + reducedValue : reducedValue; + + let canonicalRowId: string; + if (adoptTarget) { + // Pin the adopted row to local-noon only when the canonical slot + // is free or already this row — a DIFFERENT row occupying the slot + // would otherwise collide on index B. In that rare case (a tz/DST + // shift between mints) the row keeps its existing instant; the + // `stats:` externalId is the identity, measuredAt is secondary. + const slotIsFreeForTarget = + slotRow === null || slotRow.id === adoptTarget.id; + await tx.measurement.update({ + where: { id: adoptTarget.id }, + data: { + value: mergedValue, + externalId, + deletedAt: null, + ...(slotIsFreeForTarget + ? { measuredAt: canonicalTimestamp } + : {}), + }, + }); + canonicalRowId = adoptTarget.id; + } else { + const created = await tx.measurement.create({ + data: { + userId, + type, + value: reducedValue, + // pick the canonical unit from an existing row; the per-sample + // rows all carry the same unit on a given type. dayRows always + // has ≥1 row (empty buckets are skipped). + unit: + dayRows[0]?.value !== undefined + ? await resolveCanonicalUnit(tx, userId, type) + : "count", + source: "APPLE_HEALTH", + measuredAt: canonicalTimestamp, + externalId, + }, + select: { id: true }, + }); + canonicalRowId = created.id; + } + + // Hard-delete the per-sample rows that contributed to the sum. + // EXCLUDE the adopted canonical row: a per-sample row that fell on + // local-noon may be the row we just adopted as the daily total, so + // deleting it would erase the fold. `id IN (...)` is bounded by the + // per-bucket cap (largest real bucket is a ~1 440-row stepCount day). + const del = await tx.measurement.deleteMany({ + where: { id: { in: sourceRowIds, not: canonicalRowId } }, + }); + removed = del.count; }); - removed = del.count; - }); + return removed; + }; + + let removed: number; + try { + removed = await foldOnce(); + } catch (err) { + // A concurrent writer won the canonical slot mid-transaction. Retry + // once: the deterministic lookup now resolves the winning row and + // adopts it, so this pass cannot create a duplicate. Non-P2002 errors + // propagate to the per-day `onBucketError` boundary below. + if (!isUniqueConstraintViolation(err)) throw err; + removed = await foldOnce(); + } return { kind: "written", sourceRowsRemoved: removed }; }, recordBucket: ({ @@ -335,6 +385,18 @@ export async function drainPerSampleCumulative( : dayRows.length; summary.totals.dailyRowsUpserted += 1; }, + // Per-day failure boundary. A day whose write throws (e.g. a residual + // unique-index collision the adopt-in-place path could not absorb) is + // logged and STEPPED OVER so the global walk keeps draining every other + // user / type / day — the historical behaviour aborted the whole run on + // the first poisoned day (3 s die-early, 0 collapsed). + onBucketError: ({ userId, type, dateKey, error }) => { + summary.totals.daysFailed += 1; + const message = error instanceof Error ? error.message : String(error); + log( + `[drain] user=${userId} type=${type} day=${dateKey} skipped — write failed: ${message}`, + ); + }, onUserStart: ({ userId, tz, dryRun }) => { log(`[drain] user=${userId} tz=${tz}${dryRun ? " (dry-run)" : ""}`); beforeBucketsCollapsed = summary.totals.bucketsCollapsed; @@ -360,7 +422,7 @@ export async function drainPerSampleCumulative( summary.totals.usersScanned = usersScanned; log( - `[drain] done — usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`, + `[drain] done — usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFailed=${summary.totals.daysFailed}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; }