mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
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.
This commit is contained in:
1 parent
7ad0b9c4f4
commit
ddfed17fa3
6 files changed
+453
No files matched your search
@@ -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);
|
||||
@@ -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.
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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["']/);
|
||||
});
|
||||
});
|
||||
@@ -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<SleepTimelineBackfillPayload>(
|
||||
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.
|
||||
|
||||
@@ -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 (`<sleep_id>:seg:<tag>:<i>`).
|
||||
* - 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),
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user