mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 20:42:43 +00:00
Merge pull request #632 from MBombeck/release/v1.32.29
release/v1.32.29 — a re-scored Fitbit night stops double counting
This commit is contained in:
21 files changed
+1591
-62
No files matched your search
@@ -2,6 +2,35 @@
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [1.32.29] — 2026-07-25
|
||||
|
||||
Fitbit re-scores a night hours after you wake up: blocks get split, merged or
|
||||
relabelled. HealthLog identified each block by its position in the list, so a
|
||||
re-score that renumbered the blocks looked like a set of brand new ones. The
|
||||
previous version stayed on file, nothing removed it, and from then on the night
|
||||
counted twice. It was quiet: the sync kept reporting success, because the sync
|
||||
was in fact working.
|
||||
|
||||
- **A re-scored night no longer double counts.** Each sleep block is now
|
||||
identified by the moment it starts, which does not move when Fitbit rescores
|
||||
the night. A block that changes from light to deep updates in place instead
|
||||
of leaving the old copy behind, and anything the new scoring dropped is
|
||||
cleared from that night's window before the new blocks are written.
|
||||
- **Nights already affected repair themselves.** A one off pass runs once per
|
||||
connection after the update and rereads the past year of sleep, so the
|
||||
duplicates collapse without you doing anything. It covers 365 days, the same
|
||||
horizon the Fitbit history import uses. Nights older than that keep whatever
|
||||
they already hold; they cannot gain new duplicates, since Fitbit does not
|
||||
rescore year old nights. Migration `0270_fitbit_sleep_repair_marker` records
|
||||
which connections have been through the pass.
|
||||
- **Rate limits are handled instead of ignored.** When Fitbit asks for a short
|
||||
wait, the request now waits and tries again once. A long wait is refused
|
||||
outright and left to the next scheduled run, so one throttled account cannot
|
||||
hold up everybody else's.
|
||||
- **Fewer pointless writes.** Fitbit rereads the last 24 hours every hour. Rows
|
||||
that came back unchanged were rewritten anyway, which showed up as needless
|
||||
traffic for the iOS app. Unchanged rows are now left alone.
|
||||
|
||||
## [1.32.28] — 2026-07-25
|
||||
|
||||
Oura used to look back exactly one week and no further. If the sync had been
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
openapi: 3.1.0
|
||||
info:
|
||||
title: HealthLog API
|
||||
version: 1.32.28
|
||||
version: 1.32.29
|
||||
description: >-
|
||||
Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest.
|
||||
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "healthlog",
|
||||
"version": "1.32.28",
|
||||
"version": "1.32.29",
|
||||
"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",
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
-- One-shot Fitbit sleep duplicate repair (stable per-segment sleep keys).
|
||||
-- Null = repair pending; stamped once the boot-time bounded sleep re-read
|
||||
-- completes for the connection.
|
||||
ALTER TABLE "fitbit_connections" ADD COLUMN "sleep_repaired_at" TIMESTAMP(3);
|
||||
@@ -3655,6 +3655,10 @@ model FitbitConnection {
|
||||
/// 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")
|
||||
/// Set once the one-shot sleep repair re-reads the bounded sleep history and
|
||||
/// collapses the duplicates the earlier volatile segment key left behind.
|
||||
/// Null = repair pending.
|
||||
sleepRepairedAt DateTime? @map("sleep_repaired_at")
|
||||
createdAt DateTime @default(now()) @map("created_at")
|
||||
updatedAt DateTime @updatedAt @map("updated_at")
|
||||
|
||||
|
||||
+1
-1
@@ -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.32.28";
|
||||
/* @sw-version-fallback */ "v1.32.29";
|
||||
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/*`
|
||||
|
||||
@@ -2,6 +2,7 @@ import { createHash } from "node:crypto";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
FITBIT_API_BASE,
|
||||
FITBIT_RETRY_AFTER_WAIT_CAP_S,
|
||||
FITBIT_FIELD_MAP,
|
||||
FITBIT_OAUTH_SCOPE,
|
||||
exchangeCode,
|
||||
@@ -20,7 +21,7 @@ import {
|
||||
mapOxygenSaturation,
|
||||
mapRespiratoryRate,
|
||||
mapRestingHeartRate,
|
||||
mapSleepSession,
|
||||
mapSleepSessionDetailed,
|
||||
mapSteps,
|
||||
mapVo2Max,
|
||||
mapWeight,
|
||||
@@ -35,13 +36,20 @@ import {
|
||||
const CREDS = { clientId: "cid", clientSecret: "csecret" };
|
||||
|
||||
/** Stub global fetch with a queue of `{ status, body }` responses. */
|
||||
function installFetchMock(pages: Array<{ status: number; body: unknown }>) {
|
||||
function installFetchMock(
|
||||
pages: Array<{
|
||||
status: number;
|
||||
body: unknown;
|
||||
headers?: Record<string, string>;
|
||||
}>,
|
||||
) {
|
||||
let i = 0;
|
||||
const fetchMock = vi.fn(async () => {
|
||||
const page = pages[Math.min(i, pages.length - 1)]!;
|
||||
i += 1;
|
||||
return {
|
||||
status: page.status,
|
||||
headers: new Headers(page.headers ?? {}),
|
||||
json: async () => page.body,
|
||||
};
|
||||
});
|
||||
@@ -487,7 +495,8 @@ describe("sleep mapping (1.2 levels.data)", () => {
|
||||
],
|
||||
});
|
||||
expect(sessions).toHaveLength(1);
|
||||
const out = mapSleepSession(sessions[0]!);
|
||||
const mapped = mapSleepSessionDetailed(sessions[0]!);
|
||||
const out = mapped.rows;
|
||||
|
||||
// One row per segment.
|
||||
expect(out).toHaveLength(3);
|
||||
@@ -505,15 +514,123 @@ describe("sleep mapping (1.2 levels.data)", () => {
|
||||
const rem = out.find((m) => m.sleepStage === "REM")!;
|
||||
expect(rem.value).toBe(45);
|
||||
|
||||
// Every fieldTag distinct (logId anchor + stage + segment index).
|
||||
// Every fieldTag distinct: the logId anchor plus the segment's own START
|
||||
// instant — no positional index, no stage label.
|
||||
const tags = out.map((m) => m.fieldTag);
|
||||
expect(new Set(tags).size).toBe(tags.length);
|
||||
expect(deep.fieldTag).toBe("999:sleep_deep:1");
|
||||
expect(deep.fieldTag).toBe(
|
||||
`999:sleep:${new Date("2026-05-10T22:30:00.000").toISOString()}`,
|
||||
);
|
||||
|
||||
// The session window spans the earliest segment start → the latest end.
|
||||
expect(mapped.windowStart!.getTime()).toBe(
|
||||
new Date("2026-05-10T22:00:00.000").getTime(),
|
||||
);
|
||||
expect(mapped.windowEnd!.getTime()).toBe(
|
||||
new Date("2026-05-11T00:15:00.000").getTime(),
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps stable per-segment externalIds across a re-score (anchor = logId)", () => {
|
||||
// Two fetches of the SAME night. On the re-score Fitbit RE-CLASSIFIES the
|
||||
// first block (light→deep), SPLITS the second, and thereby renumbers every
|
||||
// following segment — the old positional, stage-scoped key minted fresh
|
||||
// externalIds for unchanged blocks, so the upsert inserted parallel rows
|
||||
// the night total then double-counted. The anchor + segment-start key must
|
||||
// yield the SAME externalIds for the blocks that did not move.
|
||||
const night = (
|
||||
data: Array<{ dateTime: string; level: string; seconds: number }>,
|
||||
) =>
|
||||
mapSleepSessionDetailed({
|
||||
logId: 777,
|
||||
startTime: "2026-05-10T22:00:00.000",
|
||||
endTime: "2026-05-11T06:00:00.000",
|
||||
levels: { data },
|
||||
});
|
||||
|
||||
const first = night([
|
||||
{ dateTime: "2026-05-10T22:00:00.000", level: "light", seconds: 1800 },
|
||||
{ dateTime: "2026-05-10T22:30:00.000", level: "rem", seconds: 3600 },
|
||||
{ dateTime: "2026-05-10T23:30:00.000", level: "deep", seconds: 1800 },
|
||||
]);
|
||||
const rescored = night([
|
||||
// Same start, re-classified → same key, stage updates in place.
|
||||
{ dateTime: "2026-05-10T22:00:00.000", level: "deep", seconds: 1800 },
|
||||
// Split into two halves: the first keeps the block's start key, the
|
||||
// second is a genuinely new segment. Everything after it renumbers.
|
||||
{ dateTime: "2026-05-10T22:30:00.000", level: "rem", seconds: 1800 },
|
||||
{ dateTime: "2026-05-10T23:00:00.000", level: "light", seconds: 1800 },
|
||||
{ dateTime: "2026-05-10T23:30:00.000", level: "deep", seconds: 1800 },
|
||||
]);
|
||||
|
||||
const startKey = (iso: string) =>
|
||||
`777:sleep:${new Date(iso).toISOString()}`;
|
||||
expect(first.rows.map((r) => r.fieldTag)).toEqual([
|
||||
startKey("2026-05-10T22:00:00.000"),
|
||||
startKey("2026-05-10T22:30:00.000"),
|
||||
startKey("2026-05-10T23:30:00.000"),
|
||||
]);
|
||||
// The unchanged blocks keep their keys across the re-score — overwrite,
|
||||
// never a parallel duplicate. Only the genuinely new split half is new.
|
||||
expect(rescored.rows.map((r) => r.fieldTag)).toEqual([
|
||||
startKey("2026-05-10T22:00:00.000"),
|
||||
startKey("2026-05-10T22:30:00.000"),
|
||||
startKey("2026-05-10T23:00:00.000"),
|
||||
startKey("2026-05-10T23:30:00.000"),
|
||||
]);
|
||||
// The re-classification rides the stage axis (an in-place update).
|
||||
expect(first.rows[0]!.sleepStage).toBe("CORE"); // light → shallow-NREM band
|
||||
expect(rescored.rows[0]!.sleepStage).toBe("DEEP");
|
||||
});
|
||||
|
||||
it("keys a classic (asleep / restless / in_bed) log identically", () => {
|
||||
const mapped = mapSleepSessionDetailed({
|
||||
logId: 555,
|
||||
levels: {
|
||||
data: [
|
||||
{
|
||||
dateTime: "2026-05-10T22:00:00.000",
|
||||
level: "asleep",
|
||||
seconds: 600,
|
||||
},
|
||||
{
|
||||
dateTime: "2026-05-10T22:10:00.000",
|
||||
level: "restless",
|
||||
seconds: 300,
|
||||
},
|
||||
{
|
||||
dateTime: "2026-05-10T22:15:00.000",
|
||||
level: "in_bed",
|
||||
seconds: 300,
|
||||
},
|
||||
// Unknown label — skipped, and it must not shift the other keys.
|
||||
{
|
||||
dateTime: "2026-05-10T22:20:00.000",
|
||||
level: "snoring",
|
||||
seconds: 300,
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
expect(mapped.rows.map((r) => r.fieldTag)).toEqual([
|
||||
`555:sleep:${new Date("2026-05-10T22:00:00.000").toISOString()}`,
|
||||
`555:sleep:${new Date("2026-05-10T22:10:00.000").toISOString()}`,
|
||||
`555:sleep:${new Date("2026-05-10T22:15:00.000").toISOString()}`,
|
||||
]);
|
||||
expect(mapped.rows.map((r) => r.sleepStage)).toEqual([
|
||||
"ASLEEP",
|
||||
"AWAKE",
|
||||
"IN_BED",
|
||||
]);
|
||||
});
|
||||
|
||||
it("yields nothing for a session with no parseable segments", () => {
|
||||
expect(mapSleepSession({ levels: { data: [] } })).toHaveLength(0);
|
||||
expect(mapSleepSession({})).toHaveLength(0);
|
||||
const empty = mapSleepSessionDetailed({ levels: { data: [] } });
|
||||
expect(empty.rows).toHaveLength(0);
|
||||
// A null window is what stops the sweep from issuing an unbounded delete.
|
||||
expect(empty.windowStart).toBeNull();
|
||||
expect(empty.windowEnd).toBeNull();
|
||||
expect(mapSleepSessionDetailed({}).rows).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("anchors a near-midnight segment END to the user's timezone, not the process zone", () => {
|
||||
@@ -540,7 +657,7 @@ describe("sleep mapping (1.2 levels.data)", () => {
|
||||
],
|
||||
});
|
||||
|
||||
const out = mapSleepSession(sessions[0]!, "Europe/Berlin");
|
||||
const out = mapSleepSessionDetailed(sessions[0]!, "Europe/Berlin").rows;
|
||||
expect(out).toHaveLength(1);
|
||||
// 00:30 CEST (UTC+2) → 22:30 UTC on the PRIOR civil day.
|
||||
expect(out[0]!.measuredAt.toISOString()).toBe("2026-05-10T22:30:00.000Z");
|
||||
@@ -569,10 +686,119 @@ describe("sleep mapping (1.2 levels.data)", () => {
|
||||
],
|
||||
});
|
||||
|
||||
const out = mapSleepSession(sessions[0]!, "Europe/Berlin");
|
||||
const out = mapSleepSessionDetailed(sessions[0]!, "Europe/Berlin").rows;
|
||||
expect(out).toHaveLength(1);
|
||||
// anchor = start instant ISO (UTC) → fieldTag starts with that instant.
|
||||
expect(out[0]!.fieldTag).toBe("2026-05-10T21:00:00.000Z:sleep_rem:0");
|
||||
expect(out[0]!.fieldTag).toBe(
|
||||
"2026-05-10T21:00:00.000Z:sleep:2026-05-10T22:00:00.000Z",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("fitbitGet — Retry-After on a 429", () => {
|
||||
const S = new Date("2026-05-01T00:00:00.000Z");
|
||||
const E = new Date("2026-05-10T00:00:00.000Z");
|
||||
|
||||
it("waits out a Retry-After inside the cap and retries the request once", async () => {
|
||||
vi.useFakeTimers({ toFake: ["setTimeout"] });
|
||||
try {
|
||||
const fetchMock = installFetchMock([
|
||||
{ status: 429, headers: { "retry-after": "5" }, body: {} },
|
||||
{ status: 200, body: { sleep: [] } },
|
||||
]);
|
||||
|
||||
const pending = fetchSleepRange("tok", S, E);
|
||||
|
||||
// Still parked on the delay the upstream asked for — the retry must not
|
||||
// fire early (an immediate re-request would just burn the next 429).
|
||||
await vi.advanceTimersByTimeAsync(4_999);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await expect(pending).resolves.toEqual({ sleep: [] });
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("fails fast without waiting when the delay exceeds the cap", async () => {
|
||||
// An exhausted hourly budget asks for the better part of an hour. Sitting
|
||||
// that out inside a worker slot starves every other user in the tick (and
|
||||
// a pg-boss job would be lapped by its own expiration), so the request
|
||||
// throws immediately and the next scheduled tick refetches the window.
|
||||
const fetchMock = installFetchMock([
|
||||
{
|
||||
status: 429,
|
||||
headers: { "retry-after": String(FITBIT_RETRY_AFTER_WAIT_CAP_S + 1) },
|
||||
body: {},
|
||||
},
|
||||
{ status: 200, body: { sleep: [] } },
|
||||
]);
|
||||
|
||||
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
|
||||
name: "FitbitApiError",
|
||||
classification: "transient",
|
||||
httpStatus: 429,
|
||||
retryAfterSeconds: FITBIT_RETRY_AFTER_WAIT_CAP_S + 1,
|
||||
});
|
||||
// No second attempt — and no wait.
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("throws without retrying when the header is missing", async () => {
|
||||
const fetchMock = installFetchMock([
|
||||
{ status: 429, body: {} },
|
||||
{ status: 200, body: { sleep: [] } },
|
||||
]);
|
||||
|
||||
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
|
||||
name: "FitbitApiError",
|
||||
httpStatus: 429,
|
||||
retryAfterSeconds: undefined,
|
||||
});
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("ignores the HTTP-date form of Retry-After (unknown wait → fail fast)", async () => {
|
||||
const fetchMock = installFetchMock([
|
||||
{
|
||||
status: 429,
|
||||
headers: { "retry-after": "Wed, 21 Oct 2026 07:28:00 GMT" },
|
||||
body: {},
|
||||
},
|
||||
{ status: 200, body: { sleep: [] } },
|
||||
]);
|
||||
|
||||
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
|
||||
httpStatus: 429,
|
||||
});
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("gives up after a second 429 (one retry, never a loop)", async () => {
|
||||
const fetchMock = installFetchMock([
|
||||
{ status: 429, headers: { "retry-after": "0" }, body: {} },
|
||||
{ status: 429, headers: { "retry-after": "0" }, body: {} },
|
||||
{ status: 200, body: { sleep: [] } },
|
||||
]);
|
||||
|
||||
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
|
||||
httpStatus: 429,
|
||||
});
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not retry a non-429 failure", async () => {
|
||||
const fetchMock = installFetchMock([
|
||||
{ status: 500, body: {} },
|
||||
{ status: 200, body: { sleep: [] } },
|
||||
]);
|
||||
|
||||
await expect(fetchSleepRange("tok", S, E)).rejects.toMatchObject({
|
||||
httpStatus: 500,
|
||||
});
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,341 @@
|
||||
/**
|
||||
* Pins the safety contract of `replaceStaleFitbitSleep` — the replace-by-window
|
||||
* cleanup that keeps a re-scored Fitbit night from double-counting — and the
|
||||
* end-to-end heal it produces together with the stable segment key and the
|
||||
* natural-key rescue.
|
||||
*
|
||||
* The query MUST be tightly bounded: only LIVE `FITBIT` `SLEEP_DURATION` rows,
|
||||
* only inside the session window, and NEVER a row in the fresh keep-set. A
|
||||
* session with no window is skipped entirely (nothing to clean, and an
|
||||
* unbounded delete would be data loss).
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const {
|
||||
findManyMock,
|
||||
createManyMock,
|
||||
updateMock,
|
||||
updateManyMock,
|
||||
connectionFindUniqueMock,
|
||||
fetchSleepRangeMock,
|
||||
callOrder,
|
||||
} = vi.hoisted(() => ({
|
||||
findManyMock: vi.fn(async () => [] as unknown[]),
|
||||
createManyMock: vi.fn<
|
||||
() => Promise<Array<{ id: string; type: string; measuredAt: Date }>>
|
||||
>(async () => []),
|
||||
updateMock: vi.fn(async () => ({})),
|
||||
updateManyMock: vi.fn(async () => ({ count: 0 })),
|
||||
connectionFindUniqueMock: vi.fn(async () => null as unknown),
|
||||
fetchSleepRangeMock: vi.fn(async () => ({}) as unknown),
|
||||
callOrder: [] as string[],
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
measurement: {
|
||||
findMany: findManyMock,
|
||||
createManyAndReturn: createManyMock,
|
||||
update: updateMock,
|
||||
updateMany: updateManyMock,
|
||||
},
|
||||
fitbitConnection: { findUnique: connectionFindUniqueMock },
|
||||
},
|
||||
}));
|
||||
vi.mock("@/lib/crypto", () => ({
|
||||
encrypt: vi.fn((v: string) => v),
|
||||
decrypt: vi.fn((v: string) => v),
|
||||
}));
|
||||
vi.mock("@/lib/logging/context", () => ({
|
||||
getEvent: () => ({ addWarning: vi.fn() }),
|
||||
annotate: () => {},
|
||||
}));
|
||||
vi.mock("@/lib/integrations/status", () => ({
|
||||
isReauthRequired: vi.fn(async () => false),
|
||||
recordSyncFailure: vi.fn(async () => {}),
|
||||
recordSyncSuccess: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/rollups/measurement-rollups", () => ({
|
||||
collapseToTypeDayKeys: vi.fn(() => []),
|
||||
recomputeBucketsForMeasurement: vi.fn(async () => {}),
|
||||
recomputeUserRollups: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/insights/comprehensive-generate", () => ({
|
||||
invalidateStatusInsightsForTypes: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/integrations/oauth-refresh", () => ({
|
||||
persistRotatedToken: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/arrivals/measurement-emit", () => ({
|
||||
emitInsertedMeasurementArrivals: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/daily/morning-refresh-trigger", () => ({
|
||||
maybeEnqueueMorningRefresh: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/tz/resolver", () => ({
|
||||
resolveUserTimezone: vi.fn(async () => "UTC"),
|
||||
}));
|
||||
vi.mock("../credentials", () => ({
|
||||
getUserFitbitCredentials: vi.fn(async () => null),
|
||||
}));
|
||||
vi.mock("../client", async () => {
|
||||
const actual = await vi.importActual<typeof import("../client")>("../client");
|
||||
return { ...actual, fetchSleepRange: fetchSleepRangeMock };
|
||||
});
|
||||
|
||||
import { replaceStaleFitbitSleep } from "../sync-core";
|
||||
import { syncUserSleep } from "../sync-sleep";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
callOrder.length = 0;
|
||||
findManyMock.mockReset().mockImplementation(async () => {
|
||||
callOrder.push("probe");
|
||||
return [];
|
||||
});
|
||||
createManyMock.mockReset().mockImplementation(async () => {
|
||||
callOrder.push("create");
|
||||
return [];
|
||||
});
|
||||
updateMock.mockReset().mockImplementation(async () => {
|
||||
callOrder.push("update");
|
||||
return {};
|
||||
});
|
||||
updateManyMock.mockReset().mockImplementation(async () => {
|
||||
callOrder.push("sweep");
|
||||
return { count: 0 };
|
||||
});
|
||||
});
|
||||
|
||||
describe("replaceStaleFitbitSleep — bounded replace-by-window", () => {
|
||||
it("soft-deletes only live FITBIT sleep rows in the window, excluding the fresh set", async () => {
|
||||
updateManyMock.mockImplementation(async () => {
|
||||
callOrder.push("sweep");
|
||||
return { count: 1 };
|
||||
});
|
||||
const windowStart = new Date("2026-06-01T22:30:00.000Z");
|
||||
const windowEnd = new Date("2026-06-02T06:30:00.000Z");
|
||||
const keepIds = ["777:sleep:a", "777:sleep:b"];
|
||||
|
||||
const removed = await replaceStaleFitbitSleep("user-1", [
|
||||
{ windowStart, windowEnd, keepIds },
|
||||
]);
|
||||
|
||||
expect(removed).toBe(1);
|
||||
expect(updateManyMock).toHaveBeenCalledTimes(1);
|
||||
const arg = (updateManyMock.mock.calls[0]! as unknown[])[0] as {
|
||||
where: Record<string, unknown>;
|
||||
data: Record<string, unknown>;
|
||||
};
|
||||
expect(arg.where).toEqual({
|
||||
userId: "user-1",
|
||||
source: "FITBIT",
|
||||
type: "SLEEP_DURATION",
|
||||
deletedAt: null,
|
||||
measuredAt: { gte: windowStart, lte: windowEnd },
|
||||
externalId: { notIn: keepIds },
|
||||
});
|
||||
// A soft delete — the row is tombstoned, never hard-removed.
|
||||
expect(arg.data).toHaveProperty("deletedAt");
|
||||
expect(arg.data.deletedAt).toBeInstanceOf(Date);
|
||||
});
|
||||
|
||||
it("skips a session with a null window (never issues an unbounded delete)", async () => {
|
||||
await replaceStaleFitbitSleep("user-1", [
|
||||
{ windowStart: null, windowEnd: null, keepIds: ["x"] },
|
||||
{
|
||||
windowStart: new Date("2026-06-01T22:00:00.000Z"),
|
||||
windowEnd: new Date("2026-06-02T06:00:00.000Z"),
|
||||
keepIds: [],
|
||||
},
|
||||
]);
|
||||
// Both sessions are unsafe to clean (no window, or no fresh ids to keep) —
|
||||
// neither may issue a delete.
|
||||
expect(updateManyMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("never fails the sync when the cleanup query throws", async () => {
|
||||
updateManyMock.mockRejectedValueOnce(new Error("db down"));
|
||||
await expect(
|
||||
replaceStaleFitbitSleep("user-1", [
|
||||
{
|
||||
windowStart: new Date("2026-06-01T22:00:00.000Z"),
|
||||
windowEnd: new Date("2026-06-02T06:00:00.000Z"),
|
||||
keepIds: ["777:sleep:a"],
|
||||
},
|
||||
]),
|
||||
).resolves.toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
// ── End-to-end: a re-scored night heals through the write path ──────────────
|
||||
//
|
||||
// One night stored under the OLD key format, then re-fetched after a re-score.
|
||||
// The sweep tombstones the old-keyed rows FIRST, so the upsert's natural-key
|
||||
// rescue re-keys those very rows in place — no parallel copy, no silent drop
|
||||
// against the natural unique index, and the night total stays the fresh total.
|
||||
|
||||
const SEG_A_START = "2026-06-01T22:00:00.000Z";
|
||||
const SEG_A_END = new Date("2026-06-01T22:30:00.000Z");
|
||||
const SEG_B_START = "2026-06-01T22:30:00.000Z";
|
||||
const SEG_B_END = new Date("2026-06-01T23:30:00.000Z");
|
||||
const NEW_KEY_A = `777:sleep:${SEG_A_START}`;
|
||||
const NEW_KEY_B = `777:sleep:${SEG_B_START}`;
|
||||
|
||||
const RESCORED_NIGHT = {
|
||||
sleep: [
|
||||
{
|
||||
logId: 777,
|
||||
startTime: SEG_A_START,
|
||||
endTime: SEG_B_END.toISOString(),
|
||||
levels: {
|
||||
data: [
|
||||
// Re-classified from light → deep on the SAME block.
|
||||
{ dateTime: SEG_A_START, level: "deep", seconds: 1800 },
|
||||
{ dateTime: SEG_B_START, level: "rem", seconds: 3600 },
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const SYNC_WINDOW = {
|
||||
start: new Date("2026-06-01T00:00:00.000Z"),
|
||||
end: new Date("2026-06-03T00:00:00.000Z"),
|
||||
deferRollup: true,
|
||||
};
|
||||
|
||||
function connectionAlive() {
|
||||
connectionFindUniqueMock.mockResolvedValue({
|
||||
id: "conn-1",
|
||||
fitbitUserId: "fb-1",
|
||||
accessToken: "tok",
|
||||
refreshToken: "ref",
|
||||
tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
|
||||
});
|
||||
}
|
||||
|
||||
describe("syncUserSleep — a re-scored night neither duplicates nor disappears", () => {
|
||||
beforeEach(() => {
|
||||
connectionAlive();
|
||||
fetchSleepRangeMock.mockResolvedValue(RESCORED_NIGHT);
|
||||
});
|
||||
|
||||
it("sweeps the night's window BEFORE the upsert and re-keys the tombstoned rows in place", async () => {
|
||||
updateManyMock.mockImplementation(async () => {
|
||||
callOrder.push("sweep");
|
||||
return { count: 2 };
|
||||
});
|
||||
findManyMock
|
||||
.mockImplementationOnce(async () => {
|
||||
// externalId probe: the fresh keys do not exist yet (the stored rows
|
||||
// still carry the old positional format).
|
||||
callOrder.push("probe:externalId");
|
||||
return [];
|
||||
})
|
||||
.mockImplementationOnce(async () => {
|
||||
// natural-key probe: the rows the sweep just tombstoned occupy the
|
||||
// same (type, measuredAt, sleepStage) slots.
|
||||
callOrder.push("probe:naturalKey");
|
||||
return [
|
||||
{
|
||||
id: "old-a",
|
||||
type: "SLEEP_DURATION",
|
||||
measuredAt: SEG_A_END,
|
||||
sleepStage: "DEEP",
|
||||
},
|
||||
{
|
||||
id: "old-b",
|
||||
type: "SLEEP_DURATION",
|
||||
measuredAt: SEG_B_END,
|
||||
sleepStage: "REM",
|
||||
},
|
||||
];
|
||||
});
|
||||
|
||||
const imported = await syncUserSleep("user-1", SYNC_WINDOW);
|
||||
|
||||
// The sweep runs first — reversing the order would tombstone the rows the
|
||||
// upsert had just written.
|
||||
expect(callOrder[0]).toBe("sweep");
|
||||
expect(callOrder.indexOf("sweep")).toBeLessThan(
|
||||
callOrder.indexOf("probe:externalId"),
|
||||
);
|
||||
|
||||
const sweepArg = (updateManyMock.mock.calls[0]! as unknown[])[0] as {
|
||||
where: Record<string, unknown>;
|
||||
};
|
||||
expect(sweepArg.where).toMatchObject({
|
||||
userId: "user-1",
|
||||
source: "FITBIT",
|
||||
type: "SLEEP_DURATION",
|
||||
deletedAt: null,
|
||||
measuredAt: {
|
||||
gte: new Date(SEG_A_START),
|
||||
lte: SEG_B_END,
|
||||
},
|
||||
externalId: { notIn: [NEW_KEY_A, NEW_KEY_B] },
|
||||
});
|
||||
|
||||
// No parallel copy: both rows are re-keyed in place, none inserted.
|
||||
expect(createManyMock).not.toHaveBeenCalled();
|
||||
expect(updateMock).toHaveBeenCalledTimes(2);
|
||||
const updates = updateMock.mock.calls.map(
|
||||
(c) =>
|
||||
(c as unknown[])[0] as {
|
||||
where: { id: string };
|
||||
data: Record<string, unknown>;
|
||||
},
|
||||
);
|
||||
expect(updates.map((u) => u.where.id)).toEqual(["old-a", "old-b"]);
|
||||
expect(updates.map((u) => u.data.externalId)).toEqual([
|
||||
NEW_KEY_A,
|
||||
NEW_KEY_B,
|
||||
]);
|
||||
for (const u of updates) {
|
||||
expect(u.data.deletedAt).toBeNull();
|
||||
}
|
||||
// The re-classification lands on the stage axis of the same row.
|
||||
expect(updates[0]!.data.sleepStage).toBe("DEEP");
|
||||
// Two segments in, two rows out — the night total is the fresh total.
|
||||
expect(imported).toBe(2);
|
||||
});
|
||||
|
||||
it("is a no-op on the second pass (idempotent — no churn, no duplicates)", async () => {
|
||||
findManyMock.mockImplementation(async () => {
|
||||
callOrder.push("probe");
|
||||
// Both rows now carry the fresh keys and the fresh payload.
|
||||
return [
|
||||
{
|
||||
id: "row-a",
|
||||
type: "SLEEP_DURATION",
|
||||
externalId: NEW_KEY_A,
|
||||
value: 30,
|
||||
unit: "minutes",
|
||||
measuredAt: SEG_A_END,
|
||||
sleepStage: "DEEP",
|
||||
deletedAt: null,
|
||||
},
|
||||
{
|
||||
id: "row-b",
|
||||
type: "SLEEP_DURATION",
|
||||
externalId: NEW_KEY_B,
|
||||
value: 60,
|
||||
unit: "minutes",
|
||||
measuredAt: SEG_B_END,
|
||||
sleepStage: "REM",
|
||||
deletedAt: null,
|
||||
},
|
||||
];
|
||||
});
|
||||
|
||||
const imported = await syncUserSleep("user-1", SYNC_WINDOW);
|
||||
|
||||
// The sweep still runs, but it excludes both fresh keys, so it removes
|
||||
// nothing; the upsert writes nothing at all.
|
||||
expect(updateManyMock).toHaveBeenCalledTimes(1);
|
||||
expect(createManyMock).not.toHaveBeenCalled();
|
||||
expect(updateMock).not.toHaveBeenCalled();
|
||||
expect(imported).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -98,6 +98,11 @@ describe("upsertFitbitMeasurements — tombstones resurrect", () => {
|
||||
id: "row-1",
|
||||
type: "ACTIVITY_STEPS",
|
||||
externalId: "stats:steps:2026-07-08",
|
||||
value: 8123,
|
||||
unit: "steps",
|
||||
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
|
||||
sleepStage: null,
|
||||
deletedAt: new Date("2026-07-09T00:00:00.000Z"),
|
||||
},
|
||||
]);
|
||||
|
||||
@@ -136,6 +141,127 @@ describe("upsertFitbitMeasurements — tombstones resurrect", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("upsertFitbitMeasurements — no-op overwrite skip", () => {
|
||||
it("skips the update entirely for an identical LIVE row (no syncVersion churn)", async () => {
|
||||
findManyMock.mockResolvedValue([
|
||||
{
|
||||
id: "row-1",
|
||||
type: "ACTIVITY_STEPS",
|
||||
externalId: "stats:steps:2026-07-08",
|
||||
value: 8123,
|
||||
unit: "steps",
|
||||
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
|
||||
sleepStage: null,
|
||||
deletedAt: null,
|
||||
},
|
||||
]);
|
||||
|
||||
const { imported } = await upsertFitbitMeasurements(
|
||||
"user-1",
|
||||
[STEPS_READING],
|
||||
{ deferRollup: true },
|
||||
);
|
||||
|
||||
// The 24 h overlap re-fetches this row hourly; an unchanged live row must
|
||||
// not be rewritten (the unconditional syncVersion bump churned the DB and
|
||||
// every iOS delta pull).
|
||||
expect(updateMock).not.toHaveBeenCalled();
|
||||
expect(createManyMock).not.toHaveBeenCalled();
|
||||
expect(imported).toBe(0);
|
||||
});
|
||||
|
||||
it("still writes when any payload field differs on a live row", async () => {
|
||||
findManyMock.mockResolvedValue([
|
||||
{
|
||||
id: "row-1",
|
||||
type: "ACTIVITY_STEPS",
|
||||
externalId: "stats:steps:2026-07-08",
|
||||
value: 7999, // stale daily total — Fitbit re-rolled the day upward
|
||||
unit: "steps",
|
||||
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
|
||||
sleepStage: null,
|
||||
deletedAt: null,
|
||||
},
|
||||
]);
|
||||
|
||||
const { imported } = await upsertFitbitMeasurements(
|
||||
"user-1",
|
||||
[STEPS_READING],
|
||||
{ deferRollup: true },
|
||||
);
|
||||
|
||||
expect(updateMock).toHaveBeenCalledTimes(1);
|
||||
expect(imported).toBe(1);
|
||||
});
|
||||
|
||||
it("writes when only the sleep stage differs (a re-classified segment)", async () => {
|
||||
// The stable segment key deliberately drops the stage, so a re-label
|
||||
// arrives on the SAME externalId — the skip must not swallow it.
|
||||
findManyMock.mockResolvedValue([
|
||||
{
|
||||
id: "row-1",
|
||||
type: "SLEEP_DURATION",
|
||||
externalId: "777:sleep:2026-07-08T05:45:00.000Z",
|
||||
value: 45,
|
||||
unit: "minutes",
|
||||
measuredAt: new Date("2026-07-08T06:30:00.000Z"),
|
||||
sleepStage: "CORE",
|
||||
deletedAt: null,
|
||||
},
|
||||
]);
|
||||
|
||||
const { imported } = await upsertFitbitMeasurements(
|
||||
"user-1",
|
||||
[
|
||||
{
|
||||
type: "SLEEP_DURATION",
|
||||
value: 45,
|
||||
unit: "minutes",
|
||||
measuredAt: new Date("2026-07-08T06:30:00.000Z"),
|
||||
externalId: "777:sleep:2026-07-08T05:45:00.000Z",
|
||||
sleepStage: "DEEP" as const,
|
||||
},
|
||||
],
|
||||
{ deferRollup: true },
|
||||
);
|
||||
|
||||
expect(updateMock).toHaveBeenCalledTimes(1);
|
||||
const upd = (updateMock.mock.calls[0] as unknown[])[0] as {
|
||||
data: Record<string, unknown>;
|
||||
};
|
||||
expect(upd.data.sleepStage).toBe("DEEP");
|
||||
expect(imported).toBe(1);
|
||||
});
|
||||
|
||||
it("an identical TOMBSTONED row ALWAYS updates — the write is the resurrection", async () => {
|
||||
findManyMock.mockResolvedValue([
|
||||
{
|
||||
id: "row-1",
|
||||
type: "ACTIVITY_STEPS",
|
||||
externalId: "stats:steps:2026-07-08",
|
||||
value: 8123,
|
||||
unit: "steps",
|
||||
measuredAt: new Date("2026-07-08T00:00:00.000Z"),
|
||||
sleepStage: null,
|
||||
deletedAt: new Date("2026-07-09T00:00:00.000Z"),
|
||||
},
|
||||
]);
|
||||
|
||||
const { imported } = await upsertFitbitMeasurements(
|
||||
"user-1",
|
||||
[STEPS_READING],
|
||||
{ deferRollup: true },
|
||||
);
|
||||
|
||||
expect(updateMock).toHaveBeenCalledTimes(1);
|
||||
const upd = (updateMock.mock.calls[0] as unknown[])[0] as {
|
||||
data: Record<string, unknown>;
|
||||
};
|
||||
expect(upd.data.deletedAt).toBeNull();
|
||||
expect(imported).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("upsertFitbitMeasurements — natural-key migration rescue", () => {
|
||||
const SLEEP_READING = {
|
||||
type: "SLEEP_DURATION",
|
||||
|
||||
+132
-26
@@ -384,12 +384,51 @@ export function fitbitDate(d: Date): string {
|
||||
return d.toISOString().slice(0, 10);
|
||||
}
|
||||
|
||||
/** GET a classic Fitbit endpoint and return the parsed JSON body. */
|
||||
async function fitbitGet(
|
||||
/**
|
||||
* Longest `Retry-After` delay (seconds) `fitbitGet` will sit out in-process
|
||||
* before retrying a rate-limited request.
|
||||
*
|
||||
* Fitbit's per-user budget resets on the hour, so a 429 raised near the reset
|
||||
* boundary carries a small `Retry-After` that a bounded wait turns into a
|
||||
* completed walk. A large value (an exhausted budget, up to a full hour) must
|
||||
* NOT be slept out: the poll cohort walks users sequentially, so a long hold
|
||||
* starves every other user in the tick, and inside a pg-boss job the default
|
||||
* expiration would lap the wait and admit a second execution. Above the cap the
|
||||
* request fails fast — the classified transient rides the ledger, the watermark
|
||||
* holds, and the next scheduled tick refetches the window.
|
||||
*/
|
||||
export const FITBIT_RETRY_AFTER_WAIT_CAP_S = 120;
|
||||
|
||||
/**
|
||||
* Read `Retry-After` in its delay-seconds form. Fitbit sends the seconds form;
|
||||
* the HTTP-date form is deliberately ignored (returns undefined → fail fast)
|
||||
* rather than parsed into a wait of unknown length.
|
||||
*/
|
||||
function parseRetryAfterSeconds(res: Response): number | undefined {
|
||||
const raw = res.headers.get("retry-after");
|
||||
if (typeof raw !== "string") return undefined;
|
||||
const trimmed = raw.trim();
|
||||
if (!/^\d+$/.test(trimmed)) return undefined;
|
||||
const seconds = Number.parseInt(trimmed, 10);
|
||||
return Number.isFinite(seconds) ? seconds : undefined;
|
||||
}
|
||||
|
||||
function waitSeconds(seconds: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, seconds * 1000));
|
||||
}
|
||||
|
||||
interface FitbitGetAttempt {
|
||||
json: unknown;
|
||||
verdict: ReturnType<typeof classifyFitbitResponse>;
|
||||
retryAfterSeconds: number | undefined;
|
||||
}
|
||||
|
||||
/** One GET attempt against a classic Fitbit endpoint, classified but not thrown. */
|
||||
async function fitbitGetOnce(
|
||||
path: string,
|
||||
accessToken: string,
|
||||
verb: string,
|
||||
): Promise<unknown> {
|
||||
): Promise<FitbitGetAttempt> {
|
||||
const start = performance.now();
|
||||
const res = await safeFetch(`${FITBIT_API_BASE}${path}`, {
|
||||
method: "GET",
|
||||
@@ -400,22 +439,58 @@ async function fitbitGet(
|
||||
});
|
||||
const json = await res.json().catch(() => null);
|
||||
const verdict = classifyFitbitResponse(res.status);
|
||||
const retryAfterSeconds =
|
||||
res.status === 429 ? parseRetryAfterSeconds(res) : undefined;
|
||||
getEvent()?.addExternalCall({
|
||||
service: "fitbit",
|
||||
method: verb,
|
||||
duration_ms: Math.round(performance.now() - start),
|
||||
status: res.status,
|
||||
error: verdict.classification === "success" ? undefined : verdict.reason,
|
||||
error:
|
||||
verdict.classification === "success"
|
||||
? undefined
|
||||
: retryAfterSeconds === undefined
|
||||
? verdict.reason
|
||||
: `${verdict.reason} retry_after=${retryAfterSeconds}`,
|
||||
});
|
||||
if (verdict.classification !== "success") {
|
||||
throw new FitbitApiError({
|
||||
return { json, verdict, retryAfterSeconds };
|
||||
}
|
||||
|
||||
/**
|
||||
* GET a classic Fitbit endpoint and return the parsed JSON body.
|
||||
*
|
||||
* Honours `Retry-After` on a 429 within `FITBIT_RETRY_AFTER_WAIT_CAP_S`: the
|
||||
* request waits out the delay and retries ONCE. A missing header, a delay above
|
||||
* the cap, or a second 429 throws exactly as an unhandled failure would — the
|
||||
* classified error carries `retryAfterSeconds` so the wide event records what
|
||||
* the upstream asked for.
|
||||
*/
|
||||
async function fitbitGet(
|
||||
path: string,
|
||||
accessToken: string,
|
||||
verb: string,
|
||||
): Promise<unknown> {
|
||||
const toError = (attempt: FitbitGetAttempt): FitbitApiError =>
|
||||
new FitbitApiError({
|
||||
verb,
|
||||
classification: verdict.classification,
|
||||
httpStatus: verdict.httpStatus,
|
||||
reason: verdict.reason,
|
||||
classification: attempt.verdict.classification,
|
||||
httpStatus: attempt.verdict.httpStatus,
|
||||
reason: attempt.verdict.reason,
|
||||
retryAfterSeconds: attempt.retryAfterSeconds,
|
||||
});
|
||||
|
||||
const first = await fitbitGetOnce(path, accessToken, verb);
|
||||
if (first.verdict.classification === "success") return first.json;
|
||||
|
||||
const wait = first.retryAfterSeconds;
|
||||
if (wait === undefined || wait > FITBIT_RETRY_AFTER_WAIT_CAP_S) {
|
||||
throw toError(first);
|
||||
}
|
||||
return json;
|
||||
|
||||
await waitSeconds(wait);
|
||||
const second = await fitbitGetOnce(path, accessToken, verb);
|
||||
if (second.verdict.classification === "success") return second.json;
|
||||
throw toError(second);
|
||||
}
|
||||
|
||||
// ─── Field → Measurement mapping ───────────────────────────────
|
||||
@@ -995,27 +1070,57 @@ function parseLocalInstant(iso: string, tz?: string): Date | null {
|
||||
}
|
||||
|
||||
/**
|
||||
* Map one Fitbit sleep session into per-SEGMENT `SLEEP_DURATION` rows. Each
|
||||
* segment's `dateTime` is its START; END = START + seconds. value = minutes.
|
||||
* Each segment carries a stage-scoped, INDEXED fieldTag so the several segments
|
||||
* of one stage stay distinct under the `(userId, type, source, externalId)`
|
||||
* dedup key. Unknown stage labels are skipped.
|
||||
* One mapped Fitbit sleep session: the stable anchor, the session's covered time
|
||||
* window (earliest segment start → latest segment end, both UTC), and the
|
||||
* per-segment rows. The window drives the sync's replace-by-window cleanup of
|
||||
* stale rows a re-score orphaned; `rows` is what gets upserted.
|
||||
*/
|
||||
export interface FitbitSleepSessionMapped {
|
||||
/** Stable session anchor — also the `<anchor>:sleep:` externalId prefix. */
|
||||
anchor: string;
|
||||
/** Earliest segment start across the session (UTC), or null if none map. */
|
||||
windowStart: Date | null;
|
||||
/** Latest segment end across the session (UTC), or null if none map. */
|
||||
windowEnd: Date | null;
|
||||
rows: FitbitMappedMeasurement[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Map one Fitbit sleep session into per-SEGMENT `SLEEP_DURATION` rows plus its
|
||||
* covered window. Each segment's `dateTime` is its START; END = START + seconds.
|
||||
* value = minutes. Unknown stage labels are skipped; a session with no parseable
|
||||
* segment yields empty `rows` (and a null window).
|
||||
*
|
||||
* Each segment's fieldTag keys off the STABLE session anchor plus the segment's
|
||||
* own START instant — `<anchor>:sleep:<segment-start>` — NOT a positional index
|
||||
* and NOT the stage label. This is the fix for the re-score duplication: the
|
||||
* index renumbered (and a stage-scoped tag changed) whenever Fitbit re-scored a
|
||||
* night after the fact, minting fresh externalIds so the upsert created parallel
|
||||
* rows the night total then double-counted, with no sweep to clear the strays. A
|
||||
* segment's start instant is stable per block, and dropping the stage from the
|
||||
* key means a mere re-classification (light→deep on the same block) UPDATES the
|
||||
* row in place rather than orphaning it. Two segments of one session cannot
|
||||
* share a start instant, so the key stays unique within the session; across
|
||||
* sessions the logId anchor separates them.
|
||||
*
|
||||
* The 1.2 sleep log emits OFFSET-LESS local wall-clock timestamps, so `tz` (the
|
||||
* user's stored zone) anchors them to the correct UTC instant — without it a
|
||||
* non-UTC user's near-midnight segment END would shift by their UTC offset and
|
||||
* could land on the wrong wake-day in the night reconstruction.
|
||||
*/
|
||||
export function mapSleepSession(
|
||||
export function mapSleepSessionDetailed(
|
||||
session: FitbitSleepSession,
|
||||
tz?: string,
|
||||
): FitbitMappedMeasurement[] {
|
||||
const data = session.levels?.data;
|
||||
if (!Array.isArray(data) || data.length === 0) return [];
|
||||
|
||||
): FitbitSleepSessionMapped {
|
||||
const anchor = sleepSessionAnchor(session, tz);
|
||||
const out: FitbitMappedMeasurement[] = [];
|
||||
let segIndex = 0;
|
||||
const data = session.levels?.data;
|
||||
const rows: FitbitMappedMeasurement[] = [];
|
||||
let windowStart: Date | null = null;
|
||||
let windowEnd: Date | null = null;
|
||||
if (!Array.isArray(data) || data.length === 0) {
|
||||
return { anchor, windowStart, windowEnd, rows };
|
||||
}
|
||||
|
||||
for (const seg of data) {
|
||||
const stage = mapFitbitSleepStage(seg.level);
|
||||
if (!stage) continue;
|
||||
@@ -1026,17 +1131,18 @@ export function mapSleepSession(
|
||||
if (!start) continue;
|
||||
const minutes = seg.seconds / 60;
|
||||
const end = new Date(start.getTime() + seg.seconds * 1000);
|
||||
out.push({
|
||||
if (!windowStart || start < windowStart) windowStart = start;
|
||||
if (!windowEnd || end > windowEnd) windowEnd = end;
|
||||
rows.push({
|
||||
type: "SLEEP_DURATION",
|
||||
value: round2(minutes),
|
||||
unit: "minutes",
|
||||
measuredAt: end,
|
||||
fieldTag: `${anchor}:sleep_${stage.toLowerCase()}:${segIndex}`,
|
||||
fieldTag: `${anchor}:sleep:${start.toISOString()}`,
|
||||
sleepStage: stage,
|
||||
});
|
||||
segIndex += 1;
|
||||
}
|
||||
return out;
|
||||
return { anchor, windowStart, windowEnd, rows };
|
||||
}
|
||||
|
||||
/** Read the sleep-session array off the 1.2 sleep-log envelope. */
|
||||
|
||||
@@ -66,10 +66,14 @@ Scope: `sleep`. `/1.2/user/-/sleep/date/{s}/{e}.json` →
|
||||
row **per segment**, `measuredAt = segment START + seconds = segment END`, value =
|
||||
seconds → minutes. The 1.2 log carries a real per-segment series, so the rows lay
|
||||
each block at its true clock time (a MEASURED timeline — rows are NOT flagged
|
||||
reconstructed). externalId = `<logId>:sleep_<stage>:<i>` (indexed so several
|
||||
segments of one stage stay distinct), so a re-scored night overwrites in place.
|
||||
The segment `dateTime` is local wall-clock ISO without an offset (the night
|
||||
belongs to the user's local clock) — parsed as local time.
|
||||
reconstructed). externalId = `<logId>:sleep:<segment-start-ISO>` — the stable
|
||||
session anchor plus the segment's own start instant, so a re-scored night
|
||||
overwrites in place. The key carried a positional index and the stage label
|
||||
until 2026-07-25; both moved on a re-score, which minted parallel rows the night
|
||||
total then double-counted, so the key now carries neither and
|
||||
`replaceStaleFitbitSleep` sweeps the night's window before the fresh set
|
||||
upserts. The segment `dateTime` is local wall-clock ISO without an offset (the
|
||||
night belongs to the user's local clock) — parsed as local time.
|
||||
|
||||
Stage map (stages logs + classic logs): `light → CORE` (Fitbit "light" ↔ Apple
|
||||
"core" shallow-NREM band), `deep → DEEP`, `rem → REM`, `wake`/`awake`/`restless →
|
||||
@@ -112,6 +116,6 @@ ladder).
|
||||
`externalId = <logId>:<fieldTag>`. Daily-summary rows: `externalId =
|
||||
<YYYY-MM-DD>:<fieldTag>`. Daily cumulative activity rows: `externalId =
|
||||
stats:<fieldTag>:<YYYY-MM-DD>` (Apple-Health overwrite shape). Sleep:
|
||||
`<logId>:sleep_<stage>:<i>`. Workouts: `(userId, source: "FITBIT", externalId)`
|
||||
`<logId>:sleep:<segment-start-ISO>`. Workouts: `(userId, source: "FITBIT", externalId)`
|
||||
on the `Workout` table. A re-fetch of the same window (daily summaries settle
|
||||
after the fact, so the incremental overlap is 24 h) overwrites in place.
|
||||
@@ -53,15 +53,27 @@ export function classifyFitbitResponse(
|
||||
* (message shape `Fitbit <verb> error: <status>` is preserved).
|
||||
*/
|
||||
export class FitbitApiError extends IntegrationApiError {
|
||||
/**
|
||||
* The `Retry-After` delay (seconds) the upstream asked for on a 429, when it
|
||||
* sent the delay-seconds form. Undefined on every other status, on a missing
|
||||
* header, and on the HTTP-date form. `fitbitGet` waits out a value inside
|
||||
* `FITBIT_RETRY_AFTER_WAIT_CAP_S` and retries once; anything larger fails
|
||||
* fast and surfaces here so the wide event records what was asked.
|
||||
*/
|
||||
readonly retryAfterSeconds: number | undefined;
|
||||
|
||||
constructor(opts: {
|
||||
verb: string;
|
||||
classification: FitbitClassification;
|
||||
httpStatus: number | undefined;
|
||||
reason: string;
|
||||
upstreamError?: string;
|
||||
retryAfterSeconds?: number;
|
||||
}) {
|
||||
super({ vendor: "Fitbit", ...opts });
|
||||
const { retryAfterSeconds, ...base } = opts;
|
||||
super({ vendor: "Fitbit", ...base });
|
||||
this.name = "FitbitApiError";
|
||||
this.retryAfterSeconds = retryAfterSeconds;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+142
-6
@@ -267,6 +267,43 @@ export function noteHardFailure(label: string): void {
|
||||
if (tracker) tracker.failures.push(label);
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-scope tally of rows the sleep replace-by-window sweep tombstoned. The
|
||||
* count is the deploy-time signal for the one-shot sleep repair (how many
|
||||
* re-score leftovers a pass actually cleared), and `syncUserSleep` returns only
|
||||
* its imported count, so the number rides an ambient tracker exactly like the
|
||||
* soft-skip / hard-fail / rollup-defer ledgers above.
|
||||
*/
|
||||
interface SleepSweepTracker {
|
||||
removed: number;
|
||||
}
|
||||
const sleepSweepStorage = new AsyncLocalStorage<SleepSweepTracker>();
|
||||
|
||||
/** Record swept sleep rows on the ambient tracker, if one is in scope. */
|
||||
export function noteSleepSwept(removed: number): void {
|
||||
const tracker = sleepSweepStorage.getStore();
|
||||
if (tracker) tracker.removed += removed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run `fn` inside a fresh hard-fail ledger + sleep-sweep scope and surface both.
|
||||
* `syncUserFitbit` owns its own scopes; this export exists for the one-shot jobs
|
||||
* (sleep repair) that drive a single resource sync directly and must still see a
|
||||
* dead token or a hard fetch/write failure before stamping their completion
|
||||
* marker. Ported from Google Health's `runWithGoogleHealthHardFailLedger`, with
|
||||
* the sweep tally added for the repair's deploy signal.
|
||||
*/
|
||||
export async function runWithFitbitHardFailLedger<T>(
|
||||
fn: () => Promise<T>,
|
||||
): Promise<{ result: T; failures: string[]; sleepRemoved: number }> {
|
||||
const tracker: HardFailTracker = { failures: [] };
|
||||
const sweep: SleepSweepTracker = { removed: 0 };
|
||||
const result = await hardFailStorage.run(tracker, () =>
|
||||
sleepSweepStorage.run(sweep, fn),
|
||||
);
|
||||
return { result, failures: tracker.failures, sleepRemoved: sweep.removed };
|
||||
}
|
||||
|
||||
/**
|
||||
* Single-source the per-resource collection-fetch error handling. A 403 on one
|
||||
* data class soft-skips it (warn + return 0) so sibling resources still sync; a
|
||||
@@ -399,6 +436,66 @@ export interface FitbitMeasurementUpsert {
|
||||
/** Chunk size for the batched `createMany` insert of fresh readings. */
|
||||
const FITBIT_CREATE_CHUNK = 500;
|
||||
|
||||
/** One just-fetched sleep session's window + the ids that must survive it. */
|
||||
export interface FitbitSleepReplaceWindow {
|
||||
/** Earliest segment start (UTC). Null → nothing to clean for this session. */
|
||||
windowStart: Date | null;
|
||||
/** Latest segment end (UTC). */
|
||||
windowEnd: Date | null;
|
||||
/** The fresh externalIds for THIS session — never soft-deleted. */
|
||||
keepIds: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace-by-window cleanup for re-scored Fitbit sleep sessions.
|
||||
*
|
||||
* Fitbit re-scores a night after the fact. Before the stable-key fix a re-fetch
|
||||
* minted fresh externalIds (the key carried a positional index and the stage
|
||||
* label) and left the prior scoring's rows LIVE, so the night total silently
|
||||
* double-counted. For each just-fetched session this soft-deletes any LIVE
|
||||
* `FITBIT` `SLEEP_DURATION` row whose `measuredAt` falls inside the session's
|
||||
* `[windowStart, windowEnd]` but was NOT re-produced by this fetch (`keepIds`
|
||||
* are this session's fresh externalIds). A scan bounded to one session's window
|
||||
* only ever touches that night's rows — a fresh row is protected by `keepIds`,
|
||||
* and any leftover (old volatile key, or a segment Fitbit dropped) is cleared.
|
||||
* Rows OUTSIDE every returned window are never touched, so a night Fitbit did
|
||||
* not re-report this tick stays intact — no data loss, only stale duplicates and
|
||||
* re-score orphans go.
|
||||
*
|
||||
* ORDERING IS LOAD-BEARING: the caller runs this BEFORE the upsert, so the
|
||||
* old-keyed rows are tombstoned first and the upsert's natural-key rescue
|
||||
* re-keys exactly those tombstones in place rather than inserting a parallel
|
||||
* copy (or losing the insert to `skipDuplicates` against the natural unique
|
||||
* index). Best-effort — a cleanup failure never fails the user's sync.
|
||||
*/
|
||||
export async function replaceStaleFitbitSleep(
|
||||
userId: string,
|
||||
sessions: FitbitSleepReplaceWindow[],
|
||||
): Promise<number> {
|
||||
let removed = 0;
|
||||
for (const s of sessions) {
|
||||
if (!s.windowStart || !s.windowEnd || s.keepIds.length === 0) continue;
|
||||
try {
|
||||
const res = await prisma.measurement.updateMany({
|
||||
where: {
|
||||
userId,
|
||||
source: "FITBIT",
|
||||
type: "SLEEP_DURATION",
|
||||
deletedAt: null,
|
||||
measuredAt: { gte: s.windowStart, lte: s.windowEnd },
|
||||
externalId: { notIn: s.keepIds },
|
||||
},
|
||||
data: { deletedAt: new Date() },
|
||||
});
|
||||
removed += res.count;
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(`Fitbit: sleep replace-by-window failed: ${err}`);
|
||||
}
|
||||
}
|
||||
noteSleepSwept(removed);
|
||||
return removed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write a batch of mapped Fitbit readings for one user and (unless deferred)
|
||||
* fold the rollup tier + invalidate status-insight caches once at the end
|
||||
@@ -453,7 +550,15 @@ export async function upsertFitbitMeasurements(
|
||||
// insert that `skipDuplicates` silently drops against the tombstone's key,
|
||||
// permanently wedging the key (see TOMBSTONES RESURRECT above).
|
||||
const externalIds = readings.map((r) => r.externalId);
|
||||
let rowByKey = new Map<string, { id: string }>();
|
||||
interface ProbedRow {
|
||||
id: string;
|
||||
value: number;
|
||||
unit: string;
|
||||
measuredAt: Date;
|
||||
sleepStage: FitbitMeasurementUpsert["sleepStage"];
|
||||
deletedAt: Date | null;
|
||||
}
|
||||
let rowByKey = new Map<string, ProbedRow>();
|
||||
try {
|
||||
const existing = await prisma.measurement.findMany({
|
||||
where: {
|
||||
@@ -461,14 +566,34 @@ export async function upsertFitbitMeasurements(
|
||||
source: "FITBIT",
|
||||
externalId: { in: externalIds },
|
||||
},
|
||||
select: { id: true, type: true, externalId: true },
|
||||
// The payload fields ride along so the update branch can skip a no-op
|
||||
// overwrite: the 24 h overlap re-fetches every recent row hourly, and an
|
||||
// unconditional write would bump `syncVersion` (iOS delta churn) on rows
|
||||
// whose values did not change.
|
||||
select: {
|
||||
id: true,
|
||||
type: true,
|
||||
externalId: true,
|
||||
value: true,
|
||||
unit: true,
|
||||
measuredAt: true,
|
||||
sleepStage: true,
|
||||
deletedAt: true,
|
||||
},
|
||||
});
|
||||
rowByKey = new Map(
|
||||
existing
|
||||
.filter((e) => e.externalId !== null)
|
||||
.map((e) => [
|
||||
`${e.type}${DEDUP_KEY_DELIM}${e.externalId}`,
|
||||
{ id: e.id },
|
||||
{
|
||||
id: e.id,
|
||||
value: e.value,
|
||||
unit: e.unit,
|
||||
measuredAt: e.measuredAt,
|
||||
sleepStage: e.sleepStage,
|
||||
deletedAt: e.deletedAt,
|
||||
},
|
||||
]),
|
||||
);
|
||||
} catch (err) {
|
||||
@@ -500,9 +625,20 @@ export async function upsertFitbitMeasurements(
|
||||
for (const r of readings) {
|
||||
const type = r.type as MeasurementType;
|
||||
const key = `${type}${DEDUP_KEY_DELIM}${r.externalId}`;
|
||||
const live = rowByKey.get(key);
|
||||
if (live) {
|
||||
toUpdate.push({ id: live.id, r });
|
||||
const existingRow = rowByKey.get(key);
|
||||
if (existingRow) {
|
||||
// Skip the no-op overwrite: a LIVE row whose payload matches the
|
||||
// re-fetched reading exactly gains nothing from a write, and the
|
||||
// unconditional `syncVersion` bump churned the DB + every iOS delta pull
|
||||
// hourly for the whole 24 h overlap window. A TOMBSTONED row always
|
||||
// updates — the write is what resurrects it (`deletedAt: null`).
|
||||
const unchanged =
|
||||
existingRow.deletedAt === null &&
|
||||
existingRow.value === r.value &&
|
||||
existingRow.unit === r.unit &&
|
||||
existingRow.measuredAt.getTime() === r.measuredAt.getTime() &&
|
||||
(existingRow.sleepStage ?? null) === (r.sleepStage ?? null);
|
||||
if (!unchanged) toUpdate.push({ id: existingRow.id, r });
|
||||
} else if (!plannedCreateKeys.has(key)) {
|
||||
plannedCreateKeys.add(key);
|
||||
toCreate.push({
|
||||
|
||||
@@ -9,10 +9,12 @@
|
||||
* real per-segment series, so the rows lay each block at its true clock time (a
|
||||
* MEASURED timeline, not reconstructed). Upserts as `source = FITBIT`.
|
||||
*
|
||||
* Each segment carries the `sleepStage` axis and an indexed fieldTag so the
|
||||
* several segments of one stage stay distinct under the dedup key. externalId =
|
||||
* `<logId>:sleep_<stage>:<i>` — a re-scored night overwrites in place. A 24 h
|
||||
* overlap covers Fitbit's after-the-fact re-score.
|
||||
* Each segment carries the `sleepStage` axis and a fieldTag keyed on the STABLE
|
||||
* session anchor plus the segment's own start — `<logId>:sleep:<segment-start>`
|
||||
* — so a re-scored night overwrites in place instead of minting parallel rows
|
||||
* the night total would then double-count. A 24 h overlap covers Fitbit's
|
||||
* after-the-fact re-score, and `replaceStaleFitbitSleep` clears anything an
|
||||
* earlier scoring left in the night's window before the fresh set upserts.
|
||||
*
|
||||
* A per-endpoint 403 soft-skips the resource — the `sleep` scope is granted
|
||||
* independently in the consent flow.
|
||||
@@ -20,18 +22,20 @@
|
||||
import {
|
||||
FITBIT_SLEEP_RANGE_DAYS,
|
||||
fetchSleepRange,
|
||||
mapSleepSession,
|
||||
mapSleepSessionDetailed,
|
||||
readSleepSessions,
|
||||
} from "./client";
|
||||
import {
|
||||
chunkDateRanges,
|
||||
getValidToken,
|
||||
handleCollectionFetchError,
|
||||
replaceStaleFitbitSleep,
|
||||
upsertFitbitMeasurements,
|
||||
} from "./sync-core";
|
||||
import type {
|
||||
FitbitMeasurementUpsert,
|
||||
FitbitResourceSyncOptions,
|
||||
FitbitSleepReplaceWindow,
|
||||
} from "./sync-core";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { resolveUserTimezone } from "@/lib/tz/resolver";
|
||||
@@ -54,6 +58,7 @@ export async function syncUserSleep(
|
||||
const windows = chunkDateRanges(start, end, FITBIT_SLEEP_RANGE_DAYS);
|
||||
|
||||
const readings: FitbitMeasurementUpsert[] = [];
|
||||
const replaceWindows: FitbitSleepReplaceWindow[] = [];
|
||||
for (const w of windows) {
|
||||
let body: unknown;
|
||||
try {
|
||||
@@ -62,7 +67,9 @@ export async function syncUserSleep(
|
||||
return handleCollectionFetchError("fetchSleep", userId, err);
|
||||
}
|
||||
for (const session of readSleepSessions(body)) {
|
||||
for (const m of mapSleepSession(session, tz)) {
|
||||
const mapped = mapSleepSessionDetailed(session, tz);
|
||||
if (mapped.rows.length === 0) continue;
|
||||
for (const m of mapped.rows) {
|
||||
readings.push({
|
||||
type: m.type,
|
||||
value: m.value,
|
||||
@@ -72,9 +79,22 @@ export async function syncUserSleep(
|
||||
sleepStage: m.sleepStage ?? null,
|
||||
});
|
||||
}
|
||||
// Clean any stale rows a prior scoring left in this night's window before
|
||||
// the fresh set upserts — so a re-scored night reads its true total rather
|
||||
// than the sum of the old and the re-scored copies.
|
||||
replaceWindows.push({
|
||||
windowStart: mapped.windowStart,
|
||||
windowEnd: mapped.windowEnd,
|
||||
keepIds: mapped.rows.map((m) => m.fieldTag),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// BEFORE the upsert: the sweep tombstones the old-keyed rows, and the upsert's
|
||||
// natural-key rescue then re-keys those very tombstones in place. Reversing
|
||||
// the order would tombstone the rows the upsert just wrote.
|
||||
const removed = await replaceStaleFitbitSleep(userId, replaceWindows);
|
||||
|
||||
const { imported, inserted } = await upsertFitbitMeasurements(
|
||||
userId,
|
||||
readings,
|
||||
@@ -93,6 +113,8 @@ export async function syncUserSleep(
|
||||
.map((r) => r.measuredAt),
|
||||
).catch(() => {});
|
||||
|
||||
annotate({ action: { name: "fitbit.sleep.sync", details: { imported } } });
|
||||
annotate({
|
||||
action: { name: "fitbit.sleep.sync", details: { imported, removed } },
|
||||
});
|
||||
return imported;
|
||||
}
|
||||
@@ -34,6 +34,7 @@ const source =
|
||||
const FITBIT_QUEUE_CONSTS = [
|
||||
"FITBIT_SYNC_QUEUE",
|
||||
"FITBIT_BACKFILL_QUEUE",
|
||||
"FITBIT_SLEEP_REPAIR_QUEUE",
|
||||
"FITBIT_OAUTH_STATE_CLEANUP_QUEUE",
|
||||
] as const;
|
||||
|
||||
@@ -68,6 +69,9 @@ describe("reminder-worker — Fitbit sync queues", () => {
|
||||
expect(source).toMatch(
|
||||
/boss\.work[\s\S]{0,200}FITBIT_BACKFILL_QUEUE[\s\S]{0,260}enqueueIntegrationBackfillAdmission/,
|
||||
);
|
||||
expect(source).toMatch(
|
||||
/boss\.work[\s\S]{0,200}FITBIT_SLEEP_REPAIR_QUEUE[\s\S]{0,260}enqueueIntegrationBackfillAdmission/,
|
||||
);
|
||||
expect(source).toMatch(
|
||||
/boss\.work[\s\S]{0,160}FITBIT_OAUTH_STATE_CLEANUP_QUEUE[\s\S]{0,160}handleFitbitOAuthStateCleanup/,
|
||||
);
|
||||
@@ -79,6 +83,13 @@ describe("reminder-worker — Fitbit sync queues", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("wires the one-shot Fitbit sleep-repair boot discovery", () => {
|
||||
expect(source).toMatch(
|
||||
/enqueueBootTimeFitbitSleepRepair\(\s*bootStaggerSecondsFor\("fitbit-sleep-repair"\)/,
|
||||
);
|
||||
expect(source).toMatch(/from\s*["']@\/lib\/jobs\/fitbit-sleep-repair["']/);
|
||||
});
|
||||
|
||||
it("imports the Fitbit sync driver + backfill exports", () => {
|
||||
expect(source).toMatch(/from\s*["']@\/lib\/fitbit\/sync["']/);
|
||||
expect(source).toMatch(/from\s*["']@\/lib\/jobs\/fitbit-backfill["']/);
|
||||
|
||||
@@ -0,0 +1,233 @@
|
||||
/**
|
||||
* Fitbit one-shot sleep duplicate-repair self-convergence tests (mocked).
|
||||
* - discovery only matches connections not yet repaired
|
||||
* (`sleepRepairedAt IS NULL`) and singleton-keys the enqueue per user;
|
||||
* - a completed pass re-reads the BOUNDED sleep history (an explicit 365-day
|
||||
* `start`, not `syncUserSleep`'s 30-day default), stamps `sleepRepairedAt`
|
||||
* so the next discovery drops the account, and NEVER touches the
|
||||
* `lastSyncedAt` sync watermark;
|
||||
* - the discovery is best-effort — errors surface through the result value.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const { prismaMock, bossSend, syncUserSleep, isReauthRequiredMock } =
|
||||
vi.hoisted(() => ({
|
||||
prismaMock: {
|
||||
fitbitConnection: { findMany: vi.fn(), update: vi.fn() },
|
||||
},
|
||||
bossSend: vi.fn(),
|
||||
syncUserSleep: vi.fn(),
|
||||
isReauthRequiredMock: vi.fn(async () => false),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
|
||||
|
||||
vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: () => ({ send: bossSend }),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/fitbit/sync-sleep", () => ({
|
||||
syncUserSleep: (...a: unknown[]) => syncUserSleep(...a),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/logging/context", () => ({
|
||||
annotate: () => {},
|
||||
getEvent: () => null,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/integrations/status", () => ({
|
||||
isReauthRequired: isReauthRequiredMock,
|
||||
recordSyncFailure: vi.fn(async () => {}),
|
||||
recordSyncSuccess: vi.fn(async () => {}),
|
||||
}));
|
||||
|
||||
// The repair module threads the REAL hard-fail ledger from
|
||||
// `@/lib/fitbit/sync-core`; its heavy transitive deps are mocked the same way
|
||||
// the Fitbit upsert suites do.
|
||||
vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() }));
|
||||
vi.mock("@/lib/rollups/measurement-rollups", () => ({
|
||||
collapseToTypeDayKeys: vi.fn(() => []),
|
||||
recomputeBucketsForMeasurement: vi.fn(async () => {}),
|
||||
recomputeUserRollups: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/insights/comprehensive-generate", () => ({
|
||||
invalidateStatusInsightsForTypes: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/integrations/oauth-refresh", () => ({
|
||||
persistRotatedToken: vi.fn(async () => {}),
|
||||
}));
|
||||
vi.mock("@/lib/fitbit/credentials", () => ({
|
||||
getUserFitbitCredentials: vi.fn(async () => null),
|
||||
}));
|
||||
vi.mock("@/lib/fitbit/client", () => ({ refreshAccessToken: vi.fn() }));
|
||||
|
||||
import {
|
||||
FITBIT_BACKFILL_DAYS,
|
||||
FITBIT_TOKEN_HARD_FAIL,
|
||||
noteHardFailure,
|
||||
noteSleepSwept,
|
||||
} from "@/lib/fitbit/sync-core";
|
||||
import {
|
||||
enqueueBootTimeFitbitSleepRepair,
|
||||
runFitbitSleepRepairForUser,
|
||||
} from "../fitbit-sleep-repair";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
isReauthRequiredMock.mockResolvedValue(false);
|
||||
});
|
||||
|
||||
describe("enqueueBootTimeFitbitSleepRepair — discovery", () => {
|
||||
it("queries only un-repaired connections and enqueues one singleton-keyed job per user", async () => {
|
||||
prismaMock.fitbitConnection.findMany.mockResolvedValue([
|
||||
{ userId: "u1" },
|
||||
{ userId: "u2" },
|
||||
]);
|
||||
bossSend.mockResolvedValue("job-id");
|
||||
|
||||
const result = await enqueueBootTimeFitbitSleepRepair();
|
||||
|
||||
expect(prismaMock.fitbitConnection.findMany).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ where: { sleepRepairedAt: null } }),
|
||||
);
|
||||
expect(result.enqueued).toBe(2);
|
||||
expect(bossSend).toHaveBeenCalledWith(
|
||||
"fitbit-sleep-repair",
|
||||
expect.objectContaining({ userId: "u1" }),
|
||||
expect.objectContaining({ singletonKey: "fitbit-sleep-repair|u1" }),
|
||||
);
|
||||
expect(bossSend).toHaveBeenCalledWith(
|
||||
"fitbit-sleep-repair",
|
||||
expect.objectContaining({ userId: "u2" }),
|
||||
expect.objectContaining({ singletonKey: "fitbit-sleep-repair|u2" }),
|
||||
);
|
||||
});
|
||||
|
||||
it("self-converges: no un-repaired connections → nothing enqueued", async () => {
|
||||
prismaMock.fitbitConnection.findMany.mockResolvedValue([]);
|
||||
|
||||
const result = await enqueueBootTimeFitbitSleepRepair();
|
||||
|
||||
expect(result.enqueued).toBe(0);
|
||||
expect(bossSend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("never throws — surfaces a discovery error through the result value", async () => {
|
||||
prismaMock.fitbitConnection.findMany.mockRejectedValue(
|
||||
new Error("db down"),
|
||||
);
|
||||
|
||||
const result = await enqueueBootTimeFitbitSleepRepair();
|
||||
|
||||
expect(result.error).toBe("db down");
|
||||
expect(result.enqueued).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("runFitbitSleepRepairForUser", () => {
|
||||
it("re-reads the bounded 365-day window (NOT the 30-day default) and stamps sleepRepairedAt", async () => {
|
||||
syncUserSleep.mockResolvedValue(42);
|
||||
prismaMock.fitbitConnection.update.mockResolvedValue({});
|
||||
|
||||
const before = Date.now();
|
||||
const { imported } = await runFitbitSleepRepairForUser("u1");
|
||||
const after = Date.now();
|
||||
|
||||
expect(imported).toBe(42);
|
||||
// The explicit lower bound is what makes the repair reach history: the
|
||||
// resource sync's own default only walks 30 days back.
|
||||
const syncOpts = syncUserSleep.mock.calls[0]![1] as { start: Date };
|
||||
expect(syncUserSleep).toHaveBeenCalledWith("u1", expect.any(Object));
|
||||
expect(FITBIT_BACKFILL_DAYS).toBe(365);
|
||||
const horizonMs = FITBIT_BACKFILL_DAYS * 24 * 60 * 60 * 1000;
|
||||
expect(syncOpts.start).toBeInstanceOf(Date);
|
||||
expect(syncOpts.start.getTime()).toBeGreaterThanOrEqual(
|
||||
before - horizonMs - 1000,
|
||||
);
|
||||
expect(syncOpts.start.getTime()).toBeLessThanOrEqual(after - horizonMs);
|
||||
|
||||
const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0] as {
|
||||
where: Record<string, unknown>;
|
||||
data: Record<string, unknown>;
|
||||
};
|
||||
expect(updateArg.where).toEqual({ userId: "u1" });
|
||||
expect(updateArg.data.sleepRepairedAt).toBeInstanceOf(Date);
|
||||
});
|
||||
|
||||
it("does NOT touch the lastSyncedAt sync watermark", async () => {
|
||||
syncUserSleep.mockResolvedValue(7);
|
||||
prismaMock.fitbitConnection.update.mockResolvedValue({});
|
||||
|
||||
await runFitbitSleepRepairForUser("u1");
|
||||
|
||||
// The single connection write stamps ONLY the repair marker — the
|
||||
// watermark stays where the incremental orchestrator left it.
|
||||
for (const call of prismaMock.fitbitConnection.update.mock.calls) {
|
||||
const { data } = call[0] as { data: Record<string, unknown> };
|
||||
expect(data).not.toHaveProperty("lastSyncedAt");
|
||||
expect(Object.keys(data)).toEqual(["sleepRepairedAt"]);
|
||||
}
|
||||
expect(prismaMock.fitbitConnection.update).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("reports the rows the replace-by-window sweep removed (the deploy signal)", async () => {
|
||||
syncUserSleep.mockImplementation(async () => {
|
||||
noteSleepSwept(9);
|
||||
return 12;
|
||||
});
|
||||
prismaMock.fitbitConnection.update.mockResolvedValue({});
|
||||
|
||||
await expect(runFitbitSleepRepairForUser("u1")).resolves.toEqual({
|
||||
imported: 12,
|
||||
removed: 9,
|
||||
});
|
||||
});
|
||||
|
||||
it("propagates a sync failure without stamping the marker (pg-boss retries)", async () => {
|
||||
syncUserSleep.mockRejectedValue(new Error("fitbit 500"));
|
||||
|
||||
await expect(runFitbitSleepRepairForUser("u1")).rejects.toThrow(
|
||||
"fitbit 500",
|
||||
);
|
||||
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("a swallowed hard failure (ledger entry) throws and does NOT stamp — pg-boss retries", async () => {
|
||||
// `syncUserSleep` swallows fetch/write hard failures into the ambient
|
||||
// ledger and still resolves a count; the repair must read the ledger, not
|
||||
// the count, before stamping.
|
||||
syncUserSleep.mockImplementation(async () => {
|
||||
noteHardFailure("fetchSleep");
|
||||
return 5;
|
||||
});
|
||||
|
||||
await expect(runFitbitSleepRepairForUser("u1")).rejects.toThrow(
|
||||
/incomplete/,
|
||||
);
|
||||
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("a dead token returns WITHOUT stamping and WITHOUT throwing (boot discovery re-enqueues)", async () => {
|
||||
syncUserSleep.mockImplementation(async () => {
|
||||
noteHardFailure(FITBIT_TOKEN_HARD_FAIL);
|
||||
return 0;
|
||||
});
|
||||
|
||||
await expect(runFitbitSleepRepairForUser("u1")).resolves.toEqual({
|
||||
imported: 0,
|
||||
removed: 0,
|
||||
});
|
||||
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("a connection parked at error_reauth returns WITHOUT running the sync or stamping", async () => {
|
||||
isReauthRequiredMock.mockResolvedValue(true);
|
||||
|
||||
await expect(runFitbitSleepRepairForUser("u1")).resolves.toEqual({
|
||||
imported: 0,
|
||||
removed: 0,
|
||||
});
|
||||
expect(syncUserSleep).not.toHaveBeenCalled();
|
||||
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -26,6 +26,7 @@ const providerSources = [
|
||||
"sleep-timeline-backfill.ts",
|
||||
"lab-biomarker-backfill.ts",
|
||||
"strava-backfill.ts",
|
||||
"fitbit-sleep-repair.ts",
|
||||
].map((file) => readFileSync(join(__dirname, "..", file), "utf8"));
|
||||
|
||||
describe("shared full-history backfill admission", () => {
|
||||
@@ -38,9 +39,10 @@ describe("shared full-history backfill admission", () => {
|
||||
"sleep-timeline-backfill",
|
||||
"lab-biomarker-backfill",
|
||||
"strava-backfill",
|
||||
"fitbit-sleep-repair",
|
||||
]);
|
||||
expect(INTEGRATION_BACKFILL_BOOT_ORDER.map(bootStaggerSecondsFor)).toEqual([
|
||||
30, 60, 90, 120, 150, 180, 210,
|
||||
30, 60, 90, 120, 150, 180, 210, 240,
|
||||
]);
|
||||
expect(INTEGRATION_BACKFILL_STAGGER_SECONDS).toBe(30);
|
||||
|
||||
@@ -57,6 +59,7 @@ describe("shared full-history backfill admission", () => {
|
||||
"sleep-timeline-backfill": "enqueueBootTimeSleepTimelineBackfill",
|
||||
"lab-biomarker-backfill": "enqueueBootTimeLabBiomarkerBackfill",
|
||||
"strava-backfill": "enqueueBootTimeStravaBackfill",
|
||||
"fitbit-sleep-repair": "enqueueBootTimeFitbitSleepRepair",
|
||||
} as const;
|
||||
for (const kind of INTEGRATION_BACKFILL_BOOT_ORDER) {
|
||||
expect(registrarSource).toMatch(
|
||||
@@ -140,6 +143,7 @@ describe("shared full-history backfill admission", () => {
|
||||
"SLEEP_TIMELINE_BACKFILL_QUEUE",
|
||||
"LAB_BIOMARKER_BACKFILL_QUEUE",
|
||||
"STRAVA_BACKFILL_QUEUE",
|
||||
"FITBIT_SLEEP_REPAIR_QUEUE",
|
||||
]) {
|
||||
expect(registrarSource).toMatch(
|
||||
new RegExp(
|
||||
|
||||
@@ -92,6 +92,7 @@ describe("integration-sync registrar queue policies", () => {
|
||||
["GOOGLE_HEALTH_BACKFILL_QUEUE", "exclusive"],
|
||||
["STRAVA_BACKFILL_QUEUE", "exclusive"],
|
||||
["GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE", "exclusive"],
|
||||
["FITBIT_SLEEP_REPAIR_QUEUE", "exclusive"],
|
||||
["SLEEP_TIMELINE_BACKFILL_QUEUE", "exclusive"],
|
||||
["LAB_BIOMARKER_BACKFILL_QUEUE", "exclusive"],
|
||||
])("%s is %s", (queue, policy) => {
|
||||
|
||||
@@ -0,0 +1,201 @@
|
||||
/**
|
||||
* pg-boss queue + boot-time one-shot sleep duplicate repair for Fitbit
|
||||
* connections. Modelled on the Google Health sleep repair
|
||||
* (`src/lib/jobs/google-health-sleep-repair.ts`) and the self-converging
|
||||
* backfill: a discovery query enqueues one job per connection not yet repaired,
|
||||
* and the pass is idempotent across reboots (the predicate
|
||||
* `sleep_repaired_at IS NULL` drops a connection once its repair finishes).
|
||||
*
|
||||
* Why: the earlier Fitbit sleep externalId carried a positional segment index
|
||||
* and the stage label, so every re-score of a night minted fresh keys, stranded
|
||||
* the previous scoring's rows, and over-counted the night total. The fix
|
||||
* (`replaceStaleFitbitSleep` plus the stable segment-start key) makes any
|
||||
* RE-READ night self-heal, and the incremental 24 h overlap heals recent nights
|
||||
* — but historical nights keep their duplicates until re-read. This job forces
|
||||
* one bounded sleep-history re-read per connection so the replace-by-window
|
||||
* collapses each night once.
|
||||
*
|
||||
* BOUNDED WALK: unlike the Google repair (whose API pages the whole history),
|
||||
* the classic Fitbit sleep endpoint caps each request at 30 days and the
|
||||
* per-user budget is tight, so the repair passes an explicit
|
||||
* `start = now − FITBIT_BACKFILL_DAYS` — the same 365-day horizon the backfill
|
||||
* documents as the recovery limit, about a dozen requests per account. Nights
|
||||
* older than that keep their old keys and any duplicates already recorded; they
|
||||
* can no longer accrue new ones, because Fitbit does not re-score year-old
|
||||
* nights. The explicit `start` is REQUIRED: `syncUserSleep`'s own default is
|
||||
* only 30 days.
|
||||
*
|
||||
* Watermark-safe by construction: `syncUserSleep` never reads or stamps
|
||||
* `lastSyncedAt` — `markSynced` is owned by the orchestrator (`syncUserFitbit`),
|
||||
* which this job does not call.
|
||||
*
|
||||
* The queue name MUST be registered in `allQueues` in
|
||||
* `src/lib/jobs/reminder/register-integration-sync.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 { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
|
||||
import { isReauthRequired } from "@/lib/integrations/status";
|
||||
import {
|
||||
FITBIT_BACKFILL_DAYS,
|
||||
FITBIT_TOKEN_HARD_FAIL,
|
||||
runWithFitbitHardFailLedger,
|
||||
} from "@/lib/fitbit/sync-core";
|
||||
import { syncUserSleep } from "@/lib/fitbit/sync-sleep";
|
||||
|
||||
export const FITBIT_SLEEP_REPAIR_QUEUE = "fitbit-sleep-repair";
|
||||
|
||||
/**
|
||||
* Serial concurrency — a repair walks a year of sleep history for one account
|
||||
* in 30-day requests against the tight per-user budget; concurrency-1 keeps it
|
||||
* from crowding the request pool, matching `FITBIT_BACKFILL_CONCURRENCY`.
|
||||
*/
|
||||
export const FITBIT_SLEEP_REPAIR_CONCURRENCY = 1;
|
||||
|
||||
export interface FitbitSleepRepairPayload {
|
||||
userId: string;
|
||||
enqueuedAt: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-user repair handler. Re-reads the bounded sleep history for one account —
|
||||
* `replaceStaleFitbitSleep` collapses each night's leftovers in the write path
|
||||
* and the natural-key rescue re-keys the survivors — then stamps
|
||||
* `sleepRepairedAt` so the discovery query drops it. Idempotent: the per-segment
|
||||
* upserts are key-stable and the replace-by-window clears stale copies, so a
|
||||
* re-run (e.g. a reboot mid-walk) converges rather than duplicating. Does NOT
|
||||
* move `lastSyncedAt`.
|
||||
*
|
||||
* Verdict-gated: `syncUserSleep` swallows fetch/write hard failures into the
|
||||
* ambient hard-fail ledger (returning a count either way), so the repair runs it
|
||||
* inside its own ledger scope — the same mechanism `syncUserFitbit` uses for its
|
||||
* cycle verdict — and stamps `sleepRepairedAt` ONLY on a clean run. A hard
|
||||
* fetch/write failure THROWS so pg-boss retries; a dead token (parked
|
||||
* connection, missing credentials, failed refresh) is NOT clean but does not
|
||||
* throw — retrying a dead grant burns the retry budget on the same no-op. It
|
||||
* returns WITHOUT stamping instead: the boot discovery re-enqueues the account
|
||||
* on the next boot, and the stamp lands after a reconnect.
|
||||
*/
|
||||
export async function runFitbitSleepRepairForUser(
|
||||
userId: string,
|
||||
): Promise<{ imported: number; removed: number }> {
|
||||
if (await isReauthRequired(userId, "fitbit")) {
|
||||
annotate({
|
||||
action: {
|
||||
name: "fitbit.sleepRepair.skipped_reauth",
|
||||
details: { imported: 0, removed: 0 },
|
||||
},
|
||||
});
|
||||
return { imported: 0, removed: 0 };
|
||||
}
|
||||
|
||||
const start = new Date(
|
||||
Date.now() - FITBIT_BACKFILL_DAYS * 24 * 60 * 60 * 1000,
|
||||
);
|
||||
|
||||
const {
|
||||
result: imported,
|
||||
failures,
|
||||
sleepRemoved: removed,
|
||||
} = await runWithFitbitHardFailLedger(() =>
|
||||
syncUserSleep(userId, { start, deferRollup: false }),
|
||||
);
|
||||
|
||||
if (failures.length > 0) {
|
||||
if (failures.every((f) => f === FITBIT_TOKEN_HARD_FAIL)) {
|
||||
// Dead token: nothing was fetched, so nothing to retry until the user
|
||||
// reconnects. No stamp — the next boot's discovery picks it back up.
|
||||
annotate({
|
||||
action: {
|
||||
name: "fitbit.sleepRepair.skipped_token",
|
||||
details: { imported, removed },
|
||||
},
|
||||
});
|
||||
return { imported, removed };
|
||||
}
|
||||
throw new Error(
|
||||
`fitbit sleep repair incomplete for user ${userId} (${failures.join(", ")}) — marker not stamped`,
|
||||
);
|
||||
}
|
||||
|
||||
await prisma.fitbitConnection.update({
|
||||
where: { userId },
|
||||
data: { sleepRepairedAt: new Date() },
|
||||
});
|
||||
|
||||
annotate({
|
||||
action: {
|
||||
name: "fitbit.sleepRepair.complete",
|
||||
details: { imported, removed },
|
||||
},
|
||||
});
|
||||
return { imported, removed };
|
||||
}
|
||||
|
||||
/**
|
||||
* Boot-time discovery. Finds every Fitbit connection not yet repaired
|
||||
* (`sleep_repaired_at IS NULL`) and enqueues one repair job per account.
|
||||
*
|
||||
* Idempotent across reboots: once a connection's repair completes,
|
||||
* `sleepRepairedAt` is set and the predicate drops it 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 the worker boot
|
||||
* never fails because of a repair miss.
|
||||
*/
|
||||
export async function enqueueBootTimeFitbitSleepRepair(
|
||||
startAfterSeconds: number = 0,
|
||||
): Promise<{
|
||||
enqueued: number;
|
||||
skipped: number;
|
||||
error: string | null;
|
||||
}> {
|
||||
const boss = getGlobalBoss();
|
||||
if (!boss) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
}
|
||||
|
||||
try {
|
||||
const connections = await prisma.fitbitConnection.findMany({
|
||||
where: { sleepRepairedAt: null },
|
||||
select: { userId: true },
|
||||
});
|
||||
|
||||
if (connections.length === 0) {
|
||||
return { enqueued: 0, skipped: 0, error: null };
|
||||
}
|
||||
|
||||
let enqueued = 0;
|
||||
let skipped = 0;
|
||||
for (const { userId } of connections) {
|
||||
const payload: FitbitSleepRepairPayload = {
|
||||
userId,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
};
|
||||
const jobId = await boss.send(
|
||||
FITBIT_SLEEP_REPAIR_QUEUE,
|
||||
payload,
|
||||
integrationBackfillSourceOptions(
|
||||
`fitbit-sleep-repair|${userId}`,
|
||||
startAfterSeconds,
|
||||
),
|
||||
);
|
||||
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),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -16,6 +16,9 @@ export const INTEGRATION_BACKFILL_BOOT_ORDER = [
|
||||
"sleep-timeline-backfill",
|
||||
"lab-biomarker-backfill",
|
||||
"strava-backfill",
|
||||
// Appended, not slotted next to `fitbit-backfill`, so every existing
|
||||
// provider's boot stagger stays exactly where it was.
|
||||
"fitbit-sleep-repair",
|
||||
] as const;
|
||||
|
||||
export type IntegrationBackfillKind =
|
||||
|
||||
@@ -3,9 +3,9 @@
|
||||
*
|
||||
* Owns the third-party-provider sync queues — Withings (fallback / activity /
|
||||
* sleep), WHOOP (recovery / sleep / workout / cycle + backfill + OAuth-state
|
||||
* cleanup), Fitbit (sync + backfill + OAuth-state cleanup), Nightscout, Polar,
|
||||
* Oura, MoodLog, and the two one-shot backfills (sleep-timeline, lab-biomarker)
|
||||
* — plus their boot-time discovery enqueues.
|
||||
* cleanup), Fitbit (sync + backfill + sleep repair + OAuth-state cleanup),
|
||||
* Nightscout, Polar, Oura, MoodLog, and the two one-shot backfills
|
||||
* (sleep-timeline, lab-biomarker) — plus their boot-time discovery enqueues.
|
||||
*
|
||||
* v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
|
||||
* cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
|
||||
@@ -42,6 +42,13 @@ import {
|
||||
enqueueBootTimeGoogleHealthSleepRepair,
|
||||
type GoogleHealthSleepRepairPayload,
|
||||
} from "@/lib/jobs/google-health-sleep-repair";
|
||||
import {
|
||||
FITBIT_SLEEP_REPAIR_QUEUE,
|
||||
FITBIT_SLEEP_REPAIR_CONCURRENCY,
|
||||
runFitbitSleepRepairForUser,
|
||||
enqueueBootTimeFitbitSleepRepair,
|
||||
type FitbitSleepRepairPayload,
|
||||
} from "@/lib/jobs/fitbit-sleep-repair";
|
||||
import {
|
||||
STRAVA_BACKFILL_QUEUE,
|
||||
STRAVA_BACKFILL_CONCURRENCY,
|
||||
@@ -285,6 +292,10 @@ const allQueues = [
|
||||
// self-converging boot backfill, and the daily OAuth-state ledger sweep.
|
||||
FITBIT_SYNC_QUEUE,
|
||||
FITBIT_BACKFILL_QUEUE,
|
||||
// One-shot Fitbit sleep duplicate repair. Discovery enqueues one bounded
|
||||
// sleep re-read per connection not yet repaired; the replace-by-window write
|
||||
// path collapses each night's leftovers. Idempotent across reboots.
|
||||
FITBIT_SLEEP_REPAIR_QUEUE,
|
||||
FITBIT_OAUTH_STATE_CLEANUP_QUEUE,
|
||||
// v1.26.0 — Google Health poll-only sync (no webhook at launch),
|
||||
// self-converging boot backfill, and the daily OAuth-state ledger sweep.
|
||||
@@ -411,6 +422,11 @@ const queuePolicies: QueuePolicyTable = {
|
||||
reason:
|
||||
"Per-connection one-shot sleep re-read; discovery drops the connection once it is marked repaired.",
|
||||
},
|
||||
[FITBIT_SLEEP_REPAIR_QUEUE]: {
|
||||
policy: "exclusive",
|
||||
reason:
|
||||
"Per-connection one-shot sleep re-read; discovery drops the connection once it is marked repaired.",
|
||||
},
|
||||
[SLEEP_TIMELINE_BACKFILL_QUEUE]: {
|
||||
policy: "exclusive",
|
||||
reason:
|
||||
@@ -484,6 +500,15 @@ export async function registerIntegrationSyncQueues(
|
||||
);
|
||||
break;
|
||||
}
|
||||
case "fitbit-sleep-repair": {
|
||||
const { imported, removed } =
|
||||
await runFitbitSleepRepairForUser(userId);
|
||||
workerLog(
|
||||
"info",
|
||||
`[fitbit-sleep-repair] user=${userId} imported=${imported} removed=${removed}`,
|
||||
);
|
||||
break;
|
||||
}
|
||||
case "sleep-timeline-backfill": {
|
||||
if (!data.provider) {
|
||||
throw new Error(
|
||||
@@ -616,6 +641,21 @@ export async function registerIntegrationSyncQueues(
|
||||
}
|
||||
},
|
||||
);
|
||||
// One-shot Fitbit sleep duplicate repair. This source queue retains the
|
||||
// per-connection singleton and forwards the watermark-safe bounded sleep
|
||||
// re-read to the shared drain.
|
||||
await boss.work<FitbitSleepRepairPayload>(
|
||||
FITBIT_SLEEP_REPAIR_QUEUE,
|
||||
{ localConcurrency: FITBIT_SLEEP_REPAIR_CONCURRENCY },
|
||||
async (jobs) => {
|
||||
for (const job of jobs) {
|
||||
await enqueueIntegrationBackfillAdmission(boss, {
|
||||
kind: "fitbit-sleep-repair",
|
||||
data: job.data,
|
||||
});
|
||||
}
|
||||
},
|
||||
);
|
||||
await boss.work<FitbitOAuthStateCleanupPayload>(
|
||||
FITBIT_OAUTH_STATE_CLEANUP_QUEUE,
|
||||
{ localConcurrency: 1 },
|
||||
@@ -923,4 +963,30 @@ export async function enqueueIntegrationSyncBootDiscovery(): Promise<void> {
|
||||
err,
|
||||
);
|
||||
}
|
||||
|
||||
// One-shot Fitbit sleep duplicate repair. Finds every connection not yet
|
||||
// repaired and enqueues one bounded sleep re-read per account; a completed
|
||||
// pass stamps `sleepRepairedAt`.
|
||||
try {
|
||||
const { enqueued, skipped, error } = await enqueueBootTimeFitbitSleepRepair(
|
||||
bootStaggerSecondsFor("fitbit-sleep-repair"),
|
||||
);
|
||||
if (error) {
|
||||
workerLog(
|
||||
"error",
|
||||
`[fitbit-sleep-repair] boot discovery failed: ${error}`,
|
||||
);
|
||||
} else {
|
||||
workerLog(
|
||||
"info",
|
||||
`[fitbit-sleep-repair] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
workerLog(
|
||||
"error",
|
||||
"[fitbit-sleep-repair] boot discovery threw an unexpected error",
|
||||
err,
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user