mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 13:01:00 +00:00
Merge w10/datacrit into release/v1.18.10
Preserve undecryptable encrypted day-log columns on unrelated edits (data-loss fix); give the cumulative drain a per-day error boundary so one colliding day no longer aborts the walk; apply a per-session statement timeout.
This commit is contained in:
commit
44ef6eb678
8 files changed
+764
-169
No files matched your search
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> = {}) {
|
||||
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',
|
||||
);
|
||||
});
|
||||
});
|
||||
+129
-38
@@ -114,13 +114,22 @@ export async function replaceSymptomLinks(
|
||||
}
|
||||
}
|
||||
|
||||
/** Decrypt the sensitive envelope fail-soft (null on missing / undecryptable). */
|
||||
function decryptSensitive(envelope: string | null): Partial<SensitiveFields> {
|
||||
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<SensitiveFields> } | { ok: false } {
|
||||
try {
|
||||
return JSON.parse(decrypt(envelope)) as Partial<SensitiveFields>;
|
||||
return {
|
||||
ok: true,
|
||||
values: JSON.parse(decrypt(envelope)) as Partial<SensitiveFields>,
|
||||
};
|
||||
} catch {
|
||||
return {};
|
||||
return { ok: false };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -259,21 +268,31 @@ async function writeDayLog(args: WriteArgs): Promise<DayLogWriteResult> {
|
||||
|
||||
// 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<DayLogWriteResult> {
|
||||
: 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<DayLogWriteResult> {
|
||||
? (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<DayLogWriteResult> {
|
||||
(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,
|
||||
};
|
||||
}
|
||||
+49
-8
@@ -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 });
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
select?: Record<string, unknown>;
|
||||
}) => {
|
||||
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<string, unknown> }) => {
|
||||
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<unknown>) =>
|
||||
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<string, unknown> }) => {
|
||||
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<unknown>) =>
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<DayWriteOutcome> => {
|
||||
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<number> => {
|
||||
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;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user