diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f7f113bf..03dff1c5f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,24 @@ ## [Unreleased] +## [1.28.39] — 2026-07-16 — Sync data-loss hardening + +Four fixes to sync paths that could lose or corrupt data in edge cases: + +- The nightly pass that folds raw cumulative samples (steps, energy, distance) + into daily totals no longer re-counts a sample you had deleted — it was the + one such pass missing the deleted-row filter its siblings all carry. +- That same pass is now safe when two runs overlap (a deploy-timing overlap, or + a manual admin trigger racing the nightly job): an advisory lock plus an + in-transaction re-read make the fold idempotent, so an overlap can no longer + permanently double a day's total. +- A re-scored Withings sleep segment (its end time shifts) now updates the + existing row in place instead of colliding with it and wedging that night — + the same re-key rescue the other providers already carry. +- Withings measurement sync now holds its position when a reading fails to + write, instead of advancing past it and stranding the reading; the next sync + retries it. + ## [1.28.38] — 2026-07-16 — Documents polish, waist reminders, bulk sharing The document vault gets a round of overdue polish. The list cards drop a diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 4651e5ab9..90c02bd51 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.28.38 + version: 1.28.39 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index 930216689..b636365ab 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.28.38", + "version": "1.28.39", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index 29bd81ba1..aab6ebc23 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.28.38"; + /* @sw-version-fallback */ "v1.28.39"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` 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 260a9c5b3..e27f1cde3 100644 --- a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts +++ b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts @@ -390,6 +390,15 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = const update = vi.fn().mockResolvedValue({ id: "stats-row" }); const create = vi.fn().mockResolvedValue({ id: "stats-row-new" }); const deleteMany = vi.fn().mockResolvedValue({ count: 2 }); + // v1.28.39 F2 — the in-tx re-read of the still-live per-sample contributors + // (only fires when an index-A `stats:` total already exists). Returns the + // two late rows so `liveIncrement` sums to the same 500 the pre-scan did on + // a clean single run. + const findMany = vi + .fn() + .mockResolvedValue([{ value: 300 }, { value: 200 }]); + // v1.28.39 F2 — the per-bucket advisory lock (`pg_advisory_xact_lock`). + const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]); // `findFirst` serves three roles inside `writeDay`: // 1. index-A lookup (filters on `externalId`) → the existing collapsed @@ -412,7 +421,10 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () = }, ); - const tx = { measurement: { update, create, deleteMany, findFirst } }; + const tx = { + $queryRaw, + measurement: { update, create, deleteMany, findFirst, findMany }, + }; return { prisma: { @@ -530,8 +542,13 @@ describe("drainPerSampleCumulative — colliding day does not abort the walk", ( return { unit: "count" }; }, ); + const findMany = vi.fn().mockResolvedValue([]); + const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]); - const tx = { measurement: { update, create, deleteMany, findFirst } }; + const tx = { + $queryRaw, + measurement: { update, create, deleteMany, findFirst, findMany }, + }; const prisma = { user: { findMany: findManyUser }, measurement: { findMany: findManyMeasurement }, @@ -585,8 +602,13 @@ describe("drainPerSampleCumulative — colliding day does not abort the walk", ( return { unit: "count" }; }, ); + const findMany = vi.fn().mockResolvedValue([]); + const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]); - const tx = { measurement: { update, create, deleteMany, findFirst } }; + const tx = { + $queryRaw, + measurement: { update, create, deleteMany, findFirst, findMany }, + }; const prisma = { user: { findMany: findManyUser }, measurement: { findMany: findManyMeasurement }, @@ -612,3 +634,122 @@ describe("drainPerSampleCumulative — colliding day does not abort the walk", ( expect(summary.totals.bucketsCollapsed).toBe(1); }); }); + +// v1.28.39 F1 — the cumulative drain was the ONLY one of the four consolidation +// drains whose scan `where` omitted `deletedAt: null`, so a soft-deleted +// non-`stats:` APPLE_HEALTH sample was re-summed into the day total AND +// hard-deleted (resurrecting a user-deleted grace-window sample and erasing the +// legacy-steps audit tombstones into a double-counted second `stats:` total). +// The scan must pin live rows only, matching the three siblings. +describe("drainPerSampleCumulative — scan excludes soft-deleted rows (F1)", () => { + function buildPrismaMock() { + const findManyUser = vi.fn(); + const findManyMeasurement = vi.fn().mockResolvedValue([]); + return { + user: { findMany: findManyUser }, + measurement: { findMany: findManyMeasurement }, + $transaction: vi.fn(async (cb: (tx: unknown) => Promise) => + cb({}), + ), + } as unknown as PrismaClient & { + user: { findMany: ReturnType }; + measurement: { findMany: ReturnType }; + }; + } + + it("pins the per-sample scan to deletedAt: null so tombstones are never re-summed or hard-deleted", async () => { + const prismaMock = buildPrismaMock(); + prismaMock.user.findMany.mockResolvedValue([ + { id: "user-1", timezone: "Europe/Berlin" }, + ]); + + await drainPerSampleCumulative(prismaMock, { log: () => {} }); + + const call = prismaMock.measurement.findMany.mock.calls[0]?.[0]; + expect(call).toBeDefined(); + expect(call.where.source).toBe("APPLE_HEALTH"); + // The fix: the scan is live-rows-only. Pre-fix this key was absent and a + // soft-deleted Apple sample fell into the sum + hard delete. + expect(call.where.deletedAt).toBeNull(); + }); +}); + +// v1.28.39 F2 — the additive late-fold merge (`eidRow.value + `) +// double-counted under two overlapping runs of the same (user, type, day): both +// scan the same live per-sample rows, the winner mints+hard-deletes, and the +// loser re-added its now-stale scan sum onto the winner's committed total → a +// permanent 2× that never self-healed. The per-bucket advisory lock serialises +// the folds and the in-tx re-read makes the serialised-second fold contribute +// zero. +describe("drainPerSampleCumulative — concurrent fold does not double-count (F2)", () => { + it("re-reads the live contributors in-tx so a second overlapping drain adds zero", async () => { + const findManyUser = vi + .fn() + .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]); + + // This run's SCAN still sees the two per-sample rows (it scanned before the + // sibling committed) summing to 500 in `reducedValue`. + const findManyMeasurement = vi.fn( + async (args: { where: { type: string } }) => { + if (args.where.type === "ACTIVITY_STEPS") { + return [ + { + id: "s-1", + type: "ACTIVITY_STEPS", + value: 300, + measuredAt: new Date("2026-05-16T20:00:00.000Z"), + externalId: "hk-uuid-1", + }, + { + id: "s-2", + type: "ACTIVITY_STEPS", + value: 200, + measuredAt: new Date("2026-05-16T21:00:00.000Z"), + externalId: "hk-uuid-2", + }, + ]; + } + return []; + }, + ); + + const update = vi.fn().mockResolvedValue({ id: "stats-row" }); + const create = vi.fn().mockResolvedValue({ id: "stats-row-new" }); + const deleteMany = vi.fn().mockResolvedValue({ count: 0 }); + // The sibling run already collapsed the day to 9000 (index-A row exists)… + const findFirst = vi.fn( + async (args: { where: Record }) => { + if ("externalId" in args.where) return { id: "stats-row", value: 9000 }; + if ("measuredAt" in args.where) return null; + return { unit: "count" }; + }, + ); + // …AND already HARD-DELETED s-1 / s-2, so the in-tx re-read of the live + // contributors returns nothing → increment 0, not the stale 500. + const findMany = vi.fn().mockResolvedValue([]); + const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]); + + const tx = { + $queryRaw, + measurement: { update, create, deleteMany, findFirst, findMany }, + }; + const prisma = { + user: { findMany: findManyUser }, + measurement: { findMany: findManyMeasurement }, + $transaction: vi.fn(async (cb: (t: typeof tx) => Promise) => + cb(tx), + ), + } as unknown as PrismaClient; + + await drainPerSampleCumulative(prisma, { log: () => {} }); + + // The advisory lock fired for the bucket. + expect($queryRaw).toHaveBeenCalledTimes(1); + // The row is adopted in place and left at 9000 — NOT 9500. Pre-fix the + // merge added the stale `reducedValue` (500) onto the committed total. + expect(create).not.toHaveBeenCalled(); + expect(update).toHaveBeenCalledTimes(1); + const updateArg = update.mock.calls[0]?.[0] as { data: { value: number } }; + expect(updateArg.data.value).toBe(9000); + }); +}); diff --git a/src/lib/measurements/drain-per-sample-cumulative.ts b/src/lib/measurements/drain-per-sample-cumulative.ts index 319b42933..bf9965844 100644 --- a/src/lib/measurements/drain-per-sample-cumulative.ts +++ b/src/lib/measurements/drain-per-sample-cumulative.ts @@ -215,14 +215,22 @@ export async function drainPerSampleCumulative( statsPrefix: DAILY_STATS_PREFIX, reduce: sumBucketValues, // v1.4.37 W7c — exclude rows inside the grace window so the nightly - // scheduled drain never collapses today's still-in-flight watch - // syncs. No `deletedAt` filter and no NOT-stats predicate here: the - // cumulative drain historically scans every row and relies on the - // in-memory bucketer to skip already-collapsed `stats:` rows. + // scheduled drain never collapses today's still-in-flight watch syncs. + // `deletedAt: null` pins the scan to LIVE rows — matching the three + // sibling drains (daily-mean :234, dense-intraday :464, legacy-steps + // :287). Without it a soft-deleted non-`stats:` APPLE_HEALTH sample was + // (a) re-summed into the day total and (b) hard-deleted below, which both + // resurrects a user-deleted sample inside the grace window AND erases the + // legacy-steps audit tombstones (double-counting them into a second + // `stats:` total). The NOT-`stats:` predicate stays OFF here: the + // cumulative drain relies on the in-memory bucketer to skip + // already-collapsed `stats:` rows, and the live `stats:` total (never + // tombstoned) is still scanned so the late-fold path can see it. buildScanWhere: ({ userId, type, cutoffAt }) => ({ userId, source: "APPLE_HEALTH", type, + deletedAt: null, ...(cutoffAt ? { measuredAt: { lt: cutoffAt } } : {}), }), writeDay: async ({ @@ -253,6 +261,25 @@ export async function drainPerSampleCumulative( const foldOnce = async (): Promise => { let removed = 0; await pc.$transaction(async (tx) => { + // Single-flight the per-(user, type, day) fold. Two overlapping + // drains — a rolling-deploy nightly overlap, or an operator's manual + // `POST /api/admin/drain-per-sample-cumulative` racing the cron — + // each scan the same live per-sample rows before either commits. + // Without a guard the second run re-adds its already-drained sum onto + // the first run's committed `stats:` total → a permanent 2× day total + // that never self-heals. The transaction-scoped advisory lock + // (released on commit/rollback, keyed on the bucket's canonical + // externalId) serialises the two folds so their read+write never + // interleave; the live re-read below then makes the serialised-second + // fold contribute exactly zero. Mirrors the per-user advisory-lock + // guard on the documents quota gate (`api/documents/inbound`). + // (`pg_advisory_xact_lock` returns void, which the client cannot + // deserialize as a column — selecting FROM it yields a plain int row.) + await tx.$queryRaw` + SELECT 1 AS locked + FROM pg_advisory_xact_lock(hashtextextended('drain-cumulative:' || ${userId} || ':' || ${externalId}, 0)) + `; + // Index-A row: already carries the target `stats:` externalId. When // present, this is the existing collapsed daily total. const eidRow = await tx.measurement.findFirst({ @@ -286,8 +313,27 @@ export async function drainPerSampleCumulative( // 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`. + // + // Concurrency guard (paired with the advisory lock above): a sibling + // fold that committed just before this one may have already merged + // some or all of these per-sample rows into `eidRow` and hard-deleted + // them. Trusting the pre-scan `reducedValue` for the increment would + // then re-add a sibling's already-folded work → the permanent 2×. + // Re-read the still-present contributors inside the locked + // transaction: whatever survives is the genuine un-folded increment + // (all of `reducedValue` on a clean single run; zero for a + // serialised-second run whose rows the winner already drained). + const liveIncrement = + eidRow !== null + ? ( + await tx.measurement.findMany({ + where: { id: { in: sourceRowIds } }, + select: { value: true }, + }) + ).reduce((sum, r) => sum + r.value, 0) + : 0; const mergedValue = - eidRow !== null ? eidRow.value + reducedValue : reducedValue; + eidRow !== null ? eidRow.value + liveIncrement : reducedValue; let canonicalRowId: string; if (adoptTarget) { diff --git a/src/lib/withings/__tests__/sync-measure-watermark.test.ts b/src/lib/withings/__tests__/sync-measure-watermark.test.ts new file mode 100644 index 000000000..57a542171 --- /dev/null +++ b/src/lib/withings/__tests__/sync-measure-watermark.test.ts @@ -0,0 +1,179 @@ +/** + * v1.28.39 F4 — Withings measure-sync watermark-hold contract. + * + * The documented status-tracking contract (see `syncUserMeasurements`) says a + * downstream measurement-upsert failure must record a sync failure rather than + * silently advancing the watermark. Pre-fix, `lastSyncedAt` + `recordSyncSuccess` + * were stamped UNCONDITIONALLY after a warn-only per-row loop, so a transient + * per-row write failure stranded that reading forever behind the 10-minute + * overlap window. These tests pin the fix: a hard row failure HOLDS the + * watermark (no stamp, no success, records a failure) so the next tick retries; + * a benign P2002 idempotent-write collision does NOT hold it; a clean run stamps + * as before. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ + prisma: { + measurement: { + findFirst: vi.fn(), + create: vi.fn(), + update: vi.fn(), + }, + withingsConnection: { + findUnique: vi.fn(), + update: vi.fn(), + }, + }, +})); + +vi.mock("@/lib/crypto", () => ({ + decrypt: vi.fn((v: string) => v), + encrypt: vi.fn((v: string) => v), +})); + +vi.mock("../client", () => ({ + fetchMeasurements: vi.fn(), + refreshAccessToken: vi.fn(), + subscribeWebhook: vi.fn(), +})); + +vi.mock("@/lib/integrations/status", async () => { + const actual = await vi.importActual< + typeof import("@/lib/integrations/status") + >("@/lib/integrations/status"); + return { + ...actual, + isReauthRequired: vi.fn().mockResolvedValue(false), + recordSyncFailure: vi.fn().mockResolvedValue(undefined), + recordSyncSuccess: vi.fn().mockResolvedValue(undefined), + }; +}); + +vi.mock("@/lib/rollups/measurement-rollups", async () => { + const actual = await vi.importActual< + typeof import("@/lib/rollups/measurement-rollups") + >("@/lib/rollups/measurement-rollups"); + return { + ...actual, + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + }; +}); + +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/cache/invalidate", () => ({ + invalidateUserDashboardSnapshot: vi.fn(), +})); + +vi.mock("@/lib/logging/context", () => ({ + getEvent: vi.fn(() => ({ + addExternalCall: vi.fn(), + addWarning: vi.fn(), + })), + annotate: vi.fn(), +})); + +import { prisma } from "@/lib/db"; +import { fetchMeasurements } from "../client"; +import { + recordSyncFailure, + recordSyncSuccess, +} from "@/lib/integrations/status"; + +import { syncUserMeasurements } from "../sync"; + +const P2002 = Object.assign(new Error("Unique constraint failed"), { + code: "P2002", +}); + +beforeEach(() => { + vi.clearAllMocks(); + // A live, non-expired connection so `getValidToken` returns the token + // without touching the refresh path. + vi.mocked(prisma.withingsConnection.findUnique).mockResolvedValue({ + id: "conn-1", + withingsUserId: "wu-1", + accessToken: "enc-access", + refreshToken: "enc-refresh", + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000), + } as never); + vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); +}); + +describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () => { + it("holds lastSyncedAt and records a failure when a row hard-fails", async () => { + vi.mocked(fetchMeasurements).mockResolvedValue([ + { + type: "WEIGHT", + value: 80, + measuredAt: new Date("2026-05-16T08:00:00Z"), + }, + { + type: "PULSE", + value: 60, + measuredAt: new Date("2026-05-16T08:05:00Z"), + }, + ] as never); + // The first write hard-fails (a transient DB error), the second succeeds. + vi.mocked(prisma.measurement.create) + .mockRejectedValueOnce(new Error("connection reset by peer")) + .mockResolvedValue({} as never); + + const imported = await syncUserMeasurements("user-1"); + + // Only the second row persisted. + expect(imported).toBe(1); + // The watermark is HELD — no stamp — so the next tick refetches the window. + expect(prisma.withingsConnection.update).not.toHaveBeenCalled(); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + // The failure is recorded so the streak + admin threshold reflect it. + expect(recordSyncFailure).toHaveBeenCalledTimes(1); + expect(recordSyncFailure).toHaveBeenCalledWith( + expect.objectContaining({ userId: "user-1", integration: "withings" }), + ); + }); + + it("does NOT hold the watermark for a benign P2002 idempotent-write collision", async () => { + vi.mocked(fetchMeasurements).mockResolvedValue([ + { + type: "WEIGHT", + value: 80, + measuredAt: new Date("2026-05-16T08:00:00Z"), + }, + ] as never); + vi.mocked(prisma.measurement.create).mockRejectedValueOnce(P2002); + + const imported = await syncUserMeasurements("user-1"); + + // P2002 means the row is already present — nothing lost — so the cycle is + // still healthy: stamp the watermark + success, record no failure. + expect(imported).toBe(0); + expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncFailure).not.toHaveBeenCalled(); + }); + + it("stamps the watermark + success on a clean run", async () => { + vi.mocked(fetchMeasurements).mockResolvedValue([ + { + type: "WEIGHT", + value: 80, + measuredAt: new Date("2026-05-16T08:00:00Z"), + }, + ] as never); + + const imported = await syncUserMeasurements("user-1"); + + expect(imported).toBe(1); + expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + expect(recordSyncFailure).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/withings/__tests__/sync-sleep.test.ts b/src/lib/withings/__tests__/sync-sleep.test.ts index 7cf59156d..f2c6b8912 100644 --- a/src/lib/withings/__tests__/sync-sleep.test.ts +++ b/src/lib/withings/__tests__/sync-sleep.test.ts @@ -322,6 +322,46 @@ describe("syncUserSleep — segment writes + idempotency", () => { }); }); + it("re-keys a re-scored segment in place via the natural key instead of colliding (F3, 0055 wedge)", async () => { + // Withings re-scores a night: the segment's END (measuredAt + the 0055 + // natural key) stays fixed while its START (externalId) shifts, same stage. + // The externalId probe MISSES (new START ⇒ new id), but the natural key + // `(userId, type, measuredAt, source, sleepStage)` is still occupied by the + // prior row. A blind create would P2002 on it (swallowed → row lost, then + // the sweep tombstones the old row → the night wedges forever). The + // natural-key rescue must UPDATE the surviving row in place, re-keying it. + installFetchMock([ + { startdate: 1715000500, enddate: 1715003600, state: 2, id: 42 }, + ]); + vi.mocked(prisma.measurement.findFirst).mockImplementation((async (args: { + where: Record; + }) => { + const where = args.where; + // externalId probe (START shifted ⇒ new id) → miss. + if ("externalId" in where) return null; + // natural-key probe (END + stage unchanged) → the surviving prior row. + if ("measuredAt" in where) return { id: "old-row" }; + return null; + }) as never); + vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + const imported = await syncUserSleep("user-1"); + expect(imported).toBe(1); + + // No collision: the survivor is re-keyed in place, never re-created. + expect(prisma.measurement.create).not.toHaveBeenCalled(); + expect(prisma.measurement.update).toHaveBeenCalledWith({ + where: { id: "old-row" }, + data: { + value: expect.any(Number), + // Re-keyed onto the shifted START-based externalId and resurrected. + externalId: "withings:sleep:user-1:42:1715000500", + deletedAt: null, + }, + }); + }); + it("stamps every row with an externalId keyed on session id + segment START", async () => { installFetchMock([ { startdate: 1715000000, enddate: 1715003600, state: 2, id: 42 }, diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index 395dac608..791e89726 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/src/lib/withings/sync-sleep.ts @@ -471,18 +471,56 @@ export async function syncUserSleep( }, }); } else { - await prisma.measurement.create({ - data: { + // Natural-key rescue — mirrors the measure path's findFirst in + // `sync.ts`. When Withings re-scores a night it can shift a segment's + // START (its externalId) while the END (`measuredAt`) and the stage + // stay put: the externalId probe above then MISSES, but the 0055 + // natural key `(userId, type, measuredAt, source, sleepStage)` + // (NULLS NOT DISTINCT) is already occupied by the prior row. A blind + // `create` P2002s on that key; the collision is swallowed by the catch + // below and the sweep then tombstones the old row — so the re-keyed + // segment is lost and, because a tombstoned row still holds the natural + // key, every later sync re-collides: the night wedges permanently (the + // "0055 erased re-keyed sleep" class). Probe the natural key + // (tombstone-inclusive — no `deletedAt` filter) and re-key that row in + // place, so a re-scored segment updates rather than colliding. + const naturalKeyRow = await prisma.measurement.findFirst({ + where: { userId, type: SLEEP_TYPE, - value: minutes, - unit: getUnitForType(SLEEP_TYPE), - measuredAt, source: "WITHINGS", + measuredAt, sleepStage: stage, - externalId, }, + select: { id: true }, }); + if (naturalKeyRow) { + await prisma.measurement.update({ + where: { id: naturalKeyRow.id }, + // Re-key the surviving row onto the new START-based externalId and + // resurrect it if the sweep had tombstoned it. `measuredAt` and + // `sleepStage` already match by construction (they ARE the natural + // key we matched on), so only value + externalId + deletedAt move. + data: { + value: minutes, + externalId, + deletedAt: null, + }, + }); + } else { + await prisma.measurement.create({ + data: { + userId, + type: SLEEP_TYPE, + value: minutes, + unit: getUnitForType(SLEEP_TYPE), + measuredAt, + source: "WITHINGS", + sleepStage: stage, + externalId, + }, + }); + } } touched.push({ type: SLEEP_TYPE, measuredAt }); imported++; diff --git a/src/lib/withings/sync.ts b/src/lib/withings/sync.ts index 144743154..52ce21cf3 100644 --- a/src/lib/withings/sync.ts +++ b/src/lib/withings/sync.ts @@ -12,7 +12,8 @@ import { subscribeWebhook, } from "./client"; import { getUserWithingsCredentials } from "./credentials"; -import { getEvent } from "@/lib/logging/context"; +import { annotate, getEvent } from "@/lib/logging/context"; +import { isP2002 } from "@/lib/prisma-errors"; import { isReauthRequired, recordSyncFailure, @@ -213,6 +214,16 @@ export async function syncUserMeasurements( // the user's sync. const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; + // v1.28.39 — hold-watermark-on-hard-failure (mirrors google-health / + // fitbit's `hardFailStorage` verdict). A per-row write that HARD-fails + // (anything but a benign P2002 idempotent-write collision) must not let + // `lastSyncedAt` advance past the unpersisted reading — the 10-min overlap + // window would otherwise never re-cover it and the reading is stranded + // forever. Track any hard failure so the watermark stamp + `recordSyncSuccess` + // below are held and the failure is recorded for the next tick to retry. + let anyRowFailed = false; + let firstRowError: unknown = null; + for (const m of measures) { const measType = m.type as MeasurementType; try { @@ -258,7 +269,22 @@ export async function syncUserMeasurements( touched.push({ type: measType, measuredAt: m.measuredAt }); imported++; } catch (err) { - getEvent()?.addWarning(`Failed to upsert measure: ${err}`); + if (isP2002(err)) { + // Benign: the composite unique index serialised a concurrent insert of + // this exact reading — the row is present, nothing is lost. Warn and + // continue WITHOUT holding the watermark (this is the expected + // idempotent-write race the findFirst+create models around). + getEvent()?.addWarning( + `Withings measure upsert hit an idempotent P2002 (row already present): ${err}`, + ); + } else { + // A genuine write failure strands this reading. Flag the cycle so the + // watermark is held and the next scheduled tick refetches the overlap + // window and retries. + anyRowFailed = true; + if (firstRowError === null) firstRowError = err; + getEvent()?.addWarning(`Failed to upsert measure: ${err}`); + } } } @@ -300,6 +326,29 @@ export async function syncUserMeasurements( ); } + // v1.28.39 — HOLD the watermark on any hard row failure. Implements the + // documented status-tracking contract above ("downstream measurement-upsert + // that fails ALL items → recordSyncFailure"), extended to a partial hard + // failure: a single hard-failed row must not advance `lastSyncedAt` past the + // reading it failed to persist, or the 10-min overlap window loses it. Record + // the failure (classified transient by default → the connection stays live + // and the next tick retries from the un-advanced watermark) and return + // WITHOUT stamping the watermark or success. The rows that DID write kept + // their rollup recompute above. + if (anyRowFailed) { + annotate({ + action: { + name: "withings.sync.watermark_held", + details: { imported }, + }, + }); + await recordWithingsSyncFailure( + userId, + firstRowError ?? new Error("Withings measure row write failed"), + ); + return imported; + } + // Update last synced timestamp await prisma.withingsConnection.update({ where: { userId },