From ddfed17fa31c95010a468eacbe6dc85bc53439c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 14 Jun 2026 23:16:33 +0200 Subject: [PATCH] feat(sleep): backfill existing WHOOP + Withings sleep rows after the stamp fix The WHOOP timeline reconstruction and the Withings END-stamp change the stored measuredAt (and, for WHOOP, the externalId), so existing rows keep the old broken shape until re-synced. An in-place update collides with the (userId, type, measuredAt, source, sleepStage) unique index, so the backfill deletes the affected SLEEP_DURATION rows for the source and re-syncs from scratch; the re-sync re-folds the rollup tier in its tail. Add a boot-time pg-boss one-shot per connection, modelled on the WHOOP / Fitbit backfills: discovery enqueues one job per (user, provider) where sleep_timeline_backfill_at is null, the per-user pass deletes + re-syncs and stamps the marker, and the pass is idempotent across reboots. The queue is registered in allQueues, wired to a handler, and enqueued at boot. --- .../migration.sql | 20 +++ prisma/schema.prisma | 14 ++ .../__tests__/sleep-timeline-backfill.test.ts | 149 +++++++++++++++ .../__tests__/sleep-timeline-queues.test.ts | 38 ++++ src/lib/jobs/reminder-worker.ts | 63 +++++++ src/lib/jobs/sleep-timeline-backfill.ts | 169 ++++++++++++++++++ 6 files changed, 453 insertions(+) create mode 100644 prisma/migrations/0172_v1171_sleep_timeline_backfill/migration.sql create mode 100644 src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts create mode 100644 src/lib/jobs/__tests__/sleep-timeline-queues.test.ts create mode 100644 src/lib/jobs/sleep-timeline-backfill.ts diff --git a/prisma/migrations/0172_v1171_sleep_timeline_backfill/migration.sql b/prisma/migrations/0172_v1171_sleep_timeline_backfill/migration.sql new file mode 100644 index 000000000..b5ea8e75b --- /dev/null +++ b/prisma/migrations/0172_v1171_sleep_timeline_backfill/migration.sql @@ -0,0 +1,20 @@ +-- v1.17.1 — one-shot sleep-timeline backfill markers. +-- +-- WHOOP's v2 API returns only per-stage sleep DURATION totals (no onset +-- timestamps), and the pre-fix mapper stamped all five stage totals on the one +-- sleep-END instant. The hypnogram then reconstructed every stage as a span +-- touching the night's right edge — the stacked-bar artefact with no clock +-- times. The fix reconstructs an ordered, contiguous per-segment timeline. +-- +-- Withings stamped each sleep segment with its START while every reader treats +-- `measured_at` as the segment END, shifting each night one segment-length +-- earlier. The fix stamps the END. +-- +-- Both fixes change the stored row shape / instant, so existing rows must be +-- re-synced. The boot-time backfill (per connection) deletes the affected +-- SLEEP_DURATION rows, re-syncs with the corrected mapper, re-folds the rollup +-- tier, and stamps `sleep_timeline_backfill_at` so the discovery query drops +-- the connection. Null = backfill not yet run; the marker makes the one-shot +-- idempotent across reboots. +ALTER TABLE "whoop_connections" ADD COLUMN "sleep_timeline_backfill_at" TIMESTAMP(3); +ALTER TABLE "withings_connections" ADD COLUMN "sleep_timeline_backfill_at" TIMESTAMP(3); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index ba56aa388..5558e34b5 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -2150,6 +2150,13 @@ model WithingsConnection { /// has never re-authed since v1.4.25 — the UI surfaces a banner /// inviting them to reconnect. scope String? @map("scope") + /// v1.17.1 — one-shot marker for the sleep-stamp backfill. Pre-fix Withings + /// sleep rows were stamped with the segment START; the reader treats + /// `measuredAt` as the END, shifting every night one segment-length earlier. + /// The boot-time backfill deletes the affected WITHINGS / SLEEP_DURATION rows + /// and re-syncs with the corrected END stamp, then sets this so the discovery + /// query drops the connection. Null = backfill not yet run. + sleepTimelineBackfillAt DateTime? @map("sleep_timeline_backfill_at") createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") @@ -2226,6 +2233,13 @@ model WhoopConnection { /// Set once the self-converging history backfill walks every collection to /// completion. Null = backfill still in flight (or never started). backfillCompletedAt DateTime? @map("backfill_completed_at") + /// v1.17.1 — one-shot marker for the sleep-timeline backfill. Pre-fix WHOOP + /// sleep rows stamped all five stage totals on the one sleep-END instant, so + /// the hypnogram stacked them on the night's right edge. The boot-time + /// backfill deletes the legacy summary rows and re-syncs the reconstructed + /// per-segment timeline, then sets this so the discovery query drops the + /// connection. Null = backfill not yet run. + sleepTimelineBackfillAt DateTime? @map("sleep_timeline_backfill_at") /// Max heart rate from the WHOOP body-measurement profile — a profile /// constant, not a time series, so it lives here rather than as a /// Measurement row. diff --git a/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts b/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts new file mode 100644 index 000000000..710f923b9 --- /dev/null +++ b/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts @@ -0,0 +1,149 @@ +/** + * v1.17.1 — sleep-timeline backfill self-convergence tests (mocked). + * - discovery only matches connections whose sleep rows predate the fix + * (`sleepTimelineBackfillAt IS NULL`), for both WHOOP and Withings; + * - a completed pass DELETES the source's SLEEP_DURATION rows, re-syncs, and + * stamps `sleepTimelineBackfillAt` so the next discovery drops the account + * (idempotent across reboots); + * - the discovery enqueue is singleton-keyed per (provider, user). + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { prismaMock, bossSend, syncUserWhoop, syncWithingsSleep } = vi.hoisted( + () => ({ + prismaMock: { + whoopConnection: { findMany: vi.fn(), update: vi.fn() }, + withingsConnection: { findMany: vi.fn(), update: vi.fn() }, + measurement: { deleteMany: vi.fn() }, + }, + bossSend: vi.fn(), + syncUserWhoop: vi.fn(), + syncWithingsSleep: vi.fn(), + }), +); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => ({ send: bossSend }), +})); + +vi.mock("@/lib/whoop/sync", () => ({ + syncUserWhoop: (...a: unknown[]) => syncUserWhoop(...a), +})); + +vi.mock("@/lib/withings/sync-sleep", () => ({ + syncUserSleep: (...a: unknown[]) => syncWithingsSleep(...a), +})); + +vi.mock("@/lib/logging/context", () => ({ + annotate: () => {}, + getEvent: () => null, +})); + +import { + enqueueBootTimeSleepTimelineBackfill, + runSleepTimelineBackfillForUser, +} from "../sleep-timeline-backfill"; + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe("enqueueBootTimeSleepTimelineBackfill — discovery", () => { + it("queries un-backfilled WHOOP + Withings connections and enqueues one job per (provider, user)", async () => { + prismaMock.whoopConnection.findMany.mockResolvedValue([{ userId: "w1" }]); + prismaMock.withingsConnection.findMany.mockResolvedValue([ + { userId: "v1" }, + ]); + bossSend.mockResolvedValue("job-id"); + + const result = await enqueueBootTimeSleepTimelineBackfill(); + + expect(prismaMock.whoopConnection.findMany).toHaveBeenCalledWith( + expect.objectContaining({ where: { sleepTimelineBackfillAt: null } }), + ); + expect(prismaMock.withingsConnection.findMany).toHaveBeenCalledWith( + expect.objectContaining({ where: { sleepTimelineBackfillAt: null } }), + ); + expect(result.enqueued).toBe(2); + expect(bossSend).toHaveBeenCalledWith( + "sleep-timeline-backfill", + expect.objectContaining({ userId: "w1", provider: "WHOOP" }), + expect.objectContaining({ + singletonKey: "sleep-timeline-backfill|WHOOP|w1", + }), + ); + expect(bossSend).toHaveBeenCalledWith( + "sleep-timeline-backfill", + expect.objectContaining({ userId: "v1", provider: "WITHINGS" }), + expect.objectContaining({ + singletonKey: "sleep-timeline-backfill|WITHINGS|v1", + }), + ); + }); + + it("self-converges: no un-backfilled connections → nothing enqueued", async () => { + prismaMock.whoopConnection.findMany.mockResolvedValue([]); + prismaMock.withingsConnection.findMany.mockResolvedValue([]); + + const result = await enqueueBootTimeSleepTimelineBackfill(); + + expect(result.enqueued).toBe(0); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("never throws — surfaces a discovery error through the result value", async () => { + prismaMock.whoopConnection.findMany.mockRejectedValue(new Error("db down")); + prismaMock.withingsConnection.findMany.mockResolvedValue([]); + + const result = await enqueueBootTimeSleepTimelineBackfill(); + + expect(result.error).toBe("db down"); + expect(result.enqueued).toBe(0); + }); +}); + +describe("runSleepTimelineBackfillForUser", () => { + it("WHOOP: deletes the source's sleep rows, full-syncs, and stamps the marker", async () => { + prismaMock.measurement.deleteMany.mockResolvedValue({ count: 5 }); + syncUserWhoop.mockResolvedValue(42); + prismaMock.whoopConnection.update.mockResolvedValue({}); + + const { deleted, imported } = await runSleepTimelineBackfillForUser( + "w1", + "WHOOP", + ); + + expect(deleted).toBe(5); + expect(imported).toBe(42); + expect(prismaMock.measurement.deleteMany).toHaveBeenCalledWith({ + where: { userId: "w1", type: "SLEEP_DURATION", source: "WHOOP" }, + }); + expect(syncUserWhoop).toHaveBeenCalledWith("w1", { fullSync: true }); + const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0]; + expect(updateArg.where).toEqual({ userId: "w1" }); + expect(updateArg.data.sleepTimelineBackfillAt).toBeInstanceOf(Date); + }); + + it("WITHINGS: deletes the source's sleep rows, re-syncs, and stamps the marker", async () => { + prismaMock.measurement.deleteMany.mockResolvedValue({ count: 3 }); + syncWithingsSleep.mockResolvedValue(7); + prismaMock.withingsConnection.update.mockResolvedValue({}); + + const { deleted, imported } = await runSleepTimelineBackfillForUser( + "v1", + "WITHINGS", + ); + + expect(deleted).toBe(3); + expect(imported).toBe(7); + expect(prismaMock.measurement.deleteMany).toHaveBeenCalledWith({ + where: { userId: "v1", type: "SLEEP_DURATION", source: "WITHINGS" }, + }); + expect(syncWithingsSleep).toHaveBeenCalledWith("v1", { fullSync: true }); + const updateArg = prismaMock.withingsConnection.update.mock.calls[0]![0]; + expect(updateArg.where).toEqual({ userId: "v1" }); + expect(updateArg.data.sleepTimelineBackfillAt).toBeInstanceOf(Date); + }); +}); diff --git a/src/lib/jobs/__tests__/sleep-timeline-queues.test.ts b/src/lib/jobs/__tests__/sleep-timeline-queues.test.ts new file mode 100644 index 000000000..2cbccc291 --- /dev/null +++ b/src/lib/jobs/__tests__/sleep-timeline-queues.test.ts @@ -0,0 +1,38 @@ +/** + * v1.17.1 — pg-boss queue registration for the one-shot sleep-timeline backfill. + * + * The reminder-worker imports heavy infrastructure, so we read the source as + * text and assert the queue constant is declared, registered in `allQueues`, + * wired to a `boss.work` handler, and enqueued at boot — the fast guard that + * catches the v1.4.37 dead-queue class (a queue declared but never registered, + * which silently never drains). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts"); +const source = readFileSync(REMINDER_WORKER_PATH, "utf8"); + +describe("reminder-worker — sleep-timeline backfill queue", () => { + it("registers the queue in allQueues (v1.4.37 dead-queue guard)", () => { + const block = /const allQueues = \[([\s\S]*?)\];/.exec(source); + expect(block).not.toBeNull(); + expect(block![1]!).toContain("SLEEP_TIMELINE_BACKFILL_QUEUE"); + }); + + it("wires a boss.work handler to the per-user runner", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,260}SLEEP_TIMELINE_BACKFILL_QUEUE[\s\S]{0,260}runSleepTimelineBackfillForUser/, + ); + }); + + it("wires the boot discovery", () => { + expect(source).toMatch(/enqueueBootTimeSleepTimelineBackfill\(\)/); + }); + + it("imports the backfill exports", () => { + expect(source).toMatch(/from\s*["']@\/lib\/jobs\/sleep-timeline-backfill["']/); + }); +}); diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 57aa681f0..056fbf574 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -21,6 +21,13 @@ import { enqueueBootTimeFitbitBackfill, type FitbitBackfillPayload, } from "@/lib/jobs/fitbit-backfill"; +import { + SLEEP_TIMELINE_BACKFILL_QUEUE, + SLEEP_TIMELINE_BACKFILL_CONCURRENCY, + runSleepTimelineBackfillForUser, + enqueueBootTimeSleepTimelineBackfill, + type SleepTimelineBackfillPayload, +} from "@/lib/jobs/sleep-timeline-backfill"; import { reportWorkerError } from "@/lib/jobs/report-worker-error"; import { markWorkerStarted, recordError } from "@/lib/jobs/worker-status"; import { setGlobalBoss } from "@/lib/jobs/boss-instance"; @@ -629,6 +636,13 @@ export async function startReminderWorker() { FITBIT_SYNC_QUEUE, FITBIT_BACKFILL_QUEUE, FITBIT_OAUTH_STATE_CLEANUP_QUEUE, + // v1.17.1 — one-shot sleep-timeline backfill for WHOOP + Withings. + // Discovery enqueues one job per connection whose sleep rows predate the + // stamp/shape fix; the pass deletes the affected SLEEP_DURATION rows and + // re-syncs. Idempotent across reboots. The queue MUST be registered here or + // pg-boss never provisions it and the boot enqueue silently never drains + // (the v1.4.37 dead-queue class). + SLEEP_TIMELINE_BACKFILL_QUEUE, // v1.17.0 — Nightscout CGM poll sync. Poll-only (no webhook, no OAuth, no // backfill queue — the hourly window walks the recent SGV set). The queue // MUST be registered here or pg-boss never provisions it and the schedule @@ -1056,6 +1070,27 @@ export async function startReminderWorker() { { localConcurrency: 1 }, handleFitbitOAuthStateCleanup, ); + // v1.17.1 — one-shot sleep-timeline backfill. The boot enqueue below sends + // one job per (user, provider) whose sleep rows predate the stamp/shape fix; + // this handler deletes the affected rows, re-syncs, and stamps the marker so + // the discovery query drops the connection. + await boss.work( + SLEEP_TIMELINE_BACKFILL_QUEUE, + { localConcurrency: SLEEP_TIMELINE_BACKFILL_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId, provider } = job.data; + const { deleted, imported } = await runSleepTimelineBackfillForUser( + userId, + provider, + ); + workerLog( + "info", + `[sleep-timeline-backfill] user=${userId} provider=${provider} deleted=${deleted} imported=${imported}`, + ); + } + }, + ); // v1.17.0 — Nightscout CGM poll-cohort sync. The hourly cron tick (no // `userId`) walks every configured instance; one user's unreachable host is // warned, not fatal. @@ -1882,6 +1917,34 @@ export async function startReminderWorker() { ); } + // v1.17.1 — fire-and-forget boot discovery for the one-shot sleep-timeline + // backfill. Finds every WHOOP + Withings connection whose sleep rows predate + // the stamp/shape fix and enqueues one job per (user, provider). Idempotent + // across reboots: a completed pass stamps `sleepTimelineBackfillAt`, dropping + // the connection from the discovery set. Errors come back through the + // helper's result value — the worker boot never fails because of a miss. + try { + const { enqueued, skipped, error } = + await enqueueBootTimeSleepTimelineBackfill(); + if (error) { + workerLog( + "error", + `[sleep-timeline-backfill] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[sleep-timeline-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[sleep-timeline-backfill] boot discovery threw an unexpected error", + err, + ); + } + // v1.7.0 — fire-and-forget boot discovery for the daily-mean // consolidation pass. Finds every user holding live per-sample // high-frequency mean-type rows and enqueues one job per account. diff --git a/src/lib/jobs/sleep-timeline-backfill.ts b/src/lib/jobs/sleep-timeline-backfill.ts new file mode 100644 index 000000000..d3f49f154 --- /dev/null +++ b/src/lib/jobs/sleep-timeline-backfill.ts @@ -0,0 +1,169 @@ +/** + * v1.17.1 — one-shot sleep-timeline backfill for WHOOP + Withings. + * + * Two ingest-side stamp/shape fixes changed the stored sleep rows: + * + * - WHOOP: the old mapper stamped all five per-stage DURATION totals on the + * one sleep-END instant (WHOOP v2 exposes no onset timestamps), so the + * hypnogram reconstructed every stage as a span touching the night's right + * edge — stacked, no clock times. The fix reconstructs an ordered, + * contiguous per-segment timeline with distinct `measuredAt` per segment and + * NEW indexed externalIds (`:seg::`). + * - Withings: each segment was stamped with its START while every reader + * treats `measuredAt` as the END, shifting each night one segment-length + * earlier. The fix stamps the END. + * + * Both change the row's `measuredAt` (and, for WHOOP, the externalId), and the + * unique index `(userId, type, measuredAt, source, sleepStage)` makes an + * in-place UPDATE collision-prone. So the per-connection pass DELETES the + * affected `SLEEP_DURATION` rows for that source, RE-SYNCS with the corrected + * mapper (which re-folds the rollup tier in its tail), and stamps + * `sleepTimelineBackfillAt` so the discovery query drops the connection. + * + * Idempotent across reboots: the discovery predicate is + * `sleep_timeline_backfill_at IS NULL`, and a completed pass stamps the marker. + * A reboot mid-pass re-runs delete + re-sync from scratch — the re-sync upserts + * are key-stable, so the result converges. + * + * Modelled on `whoop-backfill.ts` / `fitbit-backfill.ts`. The queue name MUST + * be registered in `allQueues` in `src/lib/jobs/reminder-worker.ts` or pg-boss + * never provisions it and the boot enqueue silently never drains (the v1.4.37 + * dead-queue class). + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { syncUserWhoop } from "@/lib/whoop/sync"; +import { syncUserSleep as syncWithingsSleep } from "@/lib/withings/sync-sleep"; + +export const SLEEP_TIMELINE_BACKFILL_QUEUE = "sleep-timeline-backfill"; + +/** + * Serial concurrency — a pass deletes a source's sleep rows then walks a + * re-sync that is rate-bounded by the upstream API cap. Concurrency-1 keeps it + * from crowding the request pool, matching the other backfill queues. + */ +export const SLEEP_TIMELINE_BACKFILL_CONCURRENCY = 1; + +export type SleepTimelineProvider = "WHOOP" | "WITHINGS"; + +export interface SleepTimelineBackfillPayload { + userId: string; + provider: SleepTimelineProvider; + enqueuedAt: string; +} + +/** + * Delete the legacy `SLEEP_DURATION` rows for one source, then re-sync with the + * corrected mapper. WHOOP runs a full-history sync (the segment externalIds are + * new, so the deleted summary rows are not re-created); Withings re-syncs its + * default trailing window with the corrected END stamp. The re-sync re-folds + * the rollup tier in its own tail. + */ +export async function runSleepTimelineBackfillForUser( + userId: string, + provider: SleepTimelineProvider, +): Promise<{ deleted: number; imported: number }> { + const { count: deleted } = await prisma.measurement.deleteMany({ + where: { userId, type: "SLEEP_DURATION", source: provider }, + }); + + let imported = 0; + if (provider === "WHOOP") { + imported = await syncUserWhoop(userId, { fullSync: true }); + await prisma.whoopConnection.update({ + where: { userId }, + data: { sleepTimelineBackfillAt: new Date() }, + }); + } else { + imported = await syncWithingsSleep(userId, { fullSync: true }); + await prisma.withingsConnection.update({ + where: { userId }, + data: { sleepTimelineBackfillAt: new Date() }, + }); + } + + annotate({ + action: { + name: "sleep.timeline.backfill.complete", + details: { provider, deleted, imported }, + }, + }); + return { deleted, imported }; +} + +/** + * Boot-time discovery. Finds every WHOOP + Withings connection not yet + * backfilled (`sleep_timeline_backfill_at IS NULL`) and enqueues one job per + * (user, provider). Idempotent across reboots: a completed pass stamps the + * marker, dropping the connection from the discovery set. pg-boss + * `singletonKey` coalesces duplicate sends so a fast restart while a job is + * queued doesn't double up. + * + * Best-effort: errors are returned through the result value so worker boot + * never fails because of a backfill miss. + */ +export async function enqueueBootTimeSleepTimelineBackfill(): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + const [whoop, withings] = await Promise.all([ + prisma.whoopConnection.findMany({ + where: { sleepTimelineBackfillAt: null }, + select: { userId: true }, + }), + prisma.withingsConnection.findMany({ + where: { sleepTimelineBackfillAt: null }, + select: { userId: true }, + }), + ]); + + const targets: Array<{ userId: string; provider: SleepTimelineProvider }> = + [ + ...whoop.map((c) => ({ userId: c.userId, provider: "WHOOP" as const })), + ...withings.map((c) => ({ + userId: c.userId, + provider: "WITHINGS" as const, + })), + ]; + + if (targets.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { userId, provider } of targets) { + const payload: SleepTimelineBackfillPayload = { + userId, + provider, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(SLEEP_TIMELINE_BACKFILL_QUEUE, payload, { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + singletonKey: `sleep-timeline-backfill|${provider}|${userId}`, + }); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +}