feat(jobs): pg-boss queues for activity-sync + sleep-v2-sync + webhook enqueue hooks

Two new pg-boss queues join the existing reminder-worker orchestrator:

  withings-activity-sync — cron '0 * * * *' (hourly at :00)
  withings-sleep-sync    — cron '15 * * * *' (hourly at :15, offset)

Each queue's handler dispatches per-user: when the payload carries
`userId` (webhook-triggered) only that user syncs; when `userId` is
absent (cron-triggered) the worker iterates every WithingsConnection.
Per-user failures log as warnings so one parked-at-reauth connection
doesn't starve the rest on a cron tick.

The webhook handler now reads the `appli` form / JSON field and
dispatches:

  appli=16  → enqueue withings-activity-sync (W17b)
  appli=44  → enqueue withings-sleep-sync    (W17c)
  appli=1/2/4 or missing → legacy syncUserMeasurements (unchanged)

When pg-boss is unavailable (dev server without the worker process)
the handler falls back to inline sync so dev iteration isn't blocked
on a separate worker process.

The 15-minute cadence offset between activity (:00) and sleep (:15)
keeps the two crons out of lockstep with the existing measure cron at
:00, smoothing the load against the Withings 120-req/min/app cap.
This commit is contained in:
Marc-André Bombeck committed 2026-05-14 18:29:59 +02:00
1 parent 6e043720ea
commit b55942fd18
5 files changed
+480 -6

No files matched your search

@@ -17,6 +17,19 @@ vi.mock("@/lib/withings/sync", () => ({
syncUserMeasurements: vi.fn(),
}));
// v1.4.25 W17b/c — activity / sleep paths dispatch through pg-boss.
vi.mock("@/lib/jobs/boss-instance", () => ({
getGlobalBoss: vi.fn(),
}));
vi.mock("@/lib/withings/sync-activity", () => ({
syncUserActivity: vi.fn(),
}));
vi.mock("@/lib/withings/sync-sleep", () => ({
syncUserSleep: vi.fn(),
}));
vi.mock("@/lib/rate-limit", () => ({
checkRateLimit: vi.fn(),
rateLimitHeaders: vi.fn(() => ({})),
@@ -17,6 +17,21 @@ vi.mock("@/lib/withings/sync", () => ({
syncUserMeasurements: vi.fn(),
}));
// v1.4.25 W17b/c — activity / sleep paths dispatch through pg-boss.
// Mock the boss-instance accessor to return a fake boss for the
// appli=16 / 44 tests and `null` otherwise.
vi.mock("@/lib/jobs/boss-instance", () => ({
getGlobalBoss: vi.fn(),
}));
vi.mock("@/lib/withings/sync-activity", () => ({
syncUserActivity: vi.fn(),
}));
vi.mock("@/lib/withings/sync-sleep", () => ({
syncUserSleep: vi.fn(),
}));
vi.mock("@/lib/rate-limit", () => ({
checkRateLimit: vi.fn(),
rateLimitHeaders: vi.fn(() => ({})),
@@ -33,8 +48,11 @@ vi.mock("@/lib/logging/context", () => ({
import { POST, GET, HEAD } from "../route";
import { prisma } from "@/lib/db";
import { syncUserMeasurements } from "@/lib/withings/sync";
import { syncUserActivity } from "@/lib/withings/sync-activity";
import { syncUserSleep } from "@/lib/withings/sync-sleep";
import { checkRateLimit } from "@/lib/rate-limit";
import { getEvent } from "@/lib/logging/context";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
const ORIGINAL_SECRET = process.env.WITHINGS_WEBHOOK_SECRET;
@@ -51,6 +69,11 @@ beforeEach(() => {
addWarning: vi.fn(),
} as never);
vi.mocked(syncUserMeasurements).mockResolvedValue(undefined as never);
// Default: pg-boss not available → activity / sleep paths fall back
// to inline sync. Individual tests override with a fake boss.
vi.mocked(getGlobalBoss).mockReturnValue(null);
vi.mocked(syncUserActivity).mockResolvedValue(0 as never);
vi.mocked(syncUserSleep).mockResolvedValue(0 as never);
});
afterEach(() => {
@@ -224,6 +247,120 @@ describe("POST /api/withings/webhook", () => {
});
});
describe("POST /api/withings/webhook — appli dispatch (W17b/c)", () => {
it("appli=16 enqueues onto withings-activity-sync (not the measure path)", async () => {
vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValueOnce({
userId: "user-act",
withingsUserId: "wu-act",
} as never);
const send = vi.fn().mockResolvedValue(undefined);
vi.mocked(getGlobalBoss).mockReturnValue({ send } as never);
const res = await POST(
formRequest(
{ userid: "wu-act", appli: "16" },
{ "x-withings-webhook-secret": "test-secret" },
),
);
expect(res.status).toBe(200);
expect(send).toHaveBeenCalledTimes(1);
expect(send.mock.calls[0][0]).toBe("withings-activity-sync");
expect(send.mock.calls[0][1]).toMatchObject({ userId: "user-act" });
// Critical: appli=16 must NOT trigger the measure path.
expect(syncUserMeasurements).not.toHaveBeenCalled();
});
it("appli=44 enqueues onto withings-sleep-sync (not the measure path)", async () => {
vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValueOnce({
userId: "user-sleep",
withingsUserId: "wu-sleep",
} as never);
const send = vi.fn().mockResolvedValue(undefined);
vi.mocked(getGlobalBoss).mockReturnValue({ send } as never);
const res = await POST(
formRequest(
{ userid: "wu-sleep", appli: "44" },
{ "x-withings-webhook-secret": "test-secret" },
),
);
expect(res.status).toBe(200);
expect(send).toHaveBeenCalledTimes(1);
expect(send.mock.calls[0][0]).toBe("withings-sleep-sync");
expect(send.mock.calls[0][1]).toMatchObject({ userId: "user-sleep" });
expect(syncUserMeasurements).not.toHaveBeenCalled();
});
it("appli=16 falls back to inline activity sync when pg-boss is unavailable", async () => {
vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValueOnce({
userId: "user-act-inline",
withingsUserId: "wu-act-inline",
} as never);
// Default beforeEach already sets `getGlobalBoss → null`.
const res = await POST(
formRequest(
{ userid: "wu-act-inline", appli: "16" },
{ "x-withings-webhook-secret": "test-secret" },
),
);
expect(res.status).toBe(200);
expect(syncUserActivity).toHaveBeenCalledWith("user-act-inline");
expect(syncUserMeasurements).not.toHaveBeenCalled();
});
it("appli=44 falls back to inline sleep sync when pg-boss is unavailable", async () => {
vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValueOnce({
userId: "user-sleep-inline",
withingsUserId: "wu-sleep-inline",
} as never);
const res = await POST(
formRequest(
{ userid: "wu-sleep-inline", appli: "44" },
{ "x-withings-webhook-secret": "test-secret" },
),
);
expect(res.status).toBe(200);
expect(syncUserSleep).toHaveBeenCalledWith("user-sleep-inline");
expect(syncUserMeasurements).not.toHaveBeenCalled();
});
it("appli=4 (BP) keeps the legacy measure path", async () => {
vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValueOnce({
userId: "user-bp",
withingsUserId: "wu-bp",
} as never);
const res = await POST(
formRequest(
{ userid: "wu-bp", appli: "4" },
{ "x-withings-webhook-secret": "test-secret" },
),
);
expect(res.status).toBe(200);
expect(syncUserMeasurements).toHaveBeenCalledWith("user-bp");
expect(syncUserActivity).not.toHaveBeenCalled();
expect(syncUserSleep).not.toHaveBeenCalled();
});
it("a missing appli falls back to the measure path (legacy Withings subscriptions)", async () => {
vi.mocked(prisma.withingsConnection.findFirst).mockResolvedValueOnce({
userId: "user-legacy",
withingsUserId: "wu-legacy",
} as never);
const res = await POST(
formRequest(
{ userid: "wu-legacy" },
{ "x-withings-webhook-secret": "test-secret" },
),
);
expect(res.status).toBe(200);
expect(syncUserMeasurements).toHaveBeenCalledWith("user-legacy");
});
});
describe("HEAD /api/withings/webhook (Withings URL verification)", () => {
it("returns 200 with a valid header secret", async () => {
const req = new NextRequest("http://localhost/api/withings/webhook", {
@@ -0,0 +1,70 @@
/**
* v1.4.25 W17b/c — pg-boss queue registration for Withings activity +
* sleep v2 syncs.
*
* The reminder-worker imports heavy infrastructure (pg-boss, Prisma
* adapter, notification dispatchers, …) so we don't boot the whole
* worker in this test. Instead we read the source file as text and
* assert the queue constants + cron schedules + handler wiring are
* present — a fast regression guard that catches accidental deletion
* of the W17b/c plumbing without spinning up Postgres.
*/
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 — Withings activity + sleep v2 queues", () => {
it("declares the activity-sync queue at the documented cadence", () => {
expect(source).toMatch(
/WITHINGS_ACTIVITY_QUEUE\s*=\s*["']withings-activity-sync["']/,
);
expect(source).toMatch(/WITHINGS_ACTIVITY_CRON\s*=\s*["']0 \* \* \* \*["']/);
});
it("declares the sleep-sync queue at the offset cadence", () => {
expect(source).toMatch(
/WITHINGS_SLEEP_QUEUE\s*=\s*["']withings-sleep-sync["']/,
);
// :15 offset keeps it out of lockstep with the measure cron at :00.
expect(source).toMatch(/WITHINGS_SLEEP_CRON\s*=\s*["']15 \* \* \* \*["']/);
});
it("registers the activity-sync handler against pg-boss", () => {
expect(source).toMatch(/handleWithingsActivitySync\b/);
expect(source).toMatch(
/boss\.work[\s\S]{0,200}WITHINGS_ACTIVITY_QUEUE[\s\S]{0,200}handleWithingsActivitySync/,
);
});
it("registers the sleep-sync handler against pg-boss", () => {
expect(source).toMatch(/handleWithingsSleepSync\b/);
expect(source).toMatch(
/boss\.work[\s\S]{0,200}WITHINGS_SLEEP_QUEUE[\s\S]{0,200}handleWithingsSleepSync/,
);
});
it("imports the new activity / sleep sync routines", () => {
expect(source).toMatch(
/import\s*\{\s*syncUserActivity\s*\}\s*from\s*["']@\/lib\/withings\/sync-activity["']/,
);
expect(source).toMatch(
/import\s*\{\s*syncUserSleep\s*\}\s*from\s*["']@\/lib\/withings\/sync-sleep["']/,
);
});
it("schedules both crons via boss.schedule (allQueues + schedules tables)", () => {
expect(source).toMatch(
/\[WITHINGS_ACTIVITY_QUEUE,\s*WITHINGS_ACTIVITY_CRON\]/,
);
expect(source).toMatch(/\[WITHINGS_SLEEP_QUEUE,\s*WITHINGS_SLEEP_CRON\]/);
});
});
+168
View File
@@ -13,6 +13,8 @@ import { PrismaPg } from "@prisma/adapter-pg";
import { dispatchNotification } from "@/lib/notifications/dispatcher";
import { parseScheduleRecurrence } from "@/lib/medication-schedule";
import { syncUserMeasurements } from "@/lib/withings/sync";
import { syncUserActivity } from "@/lib/withings/sync-activity";
import { syncUserSleep } from "@/lib/withings/sync-sleep";
import { generateGeneralStatusForUser } from "@/lib/insights/general-status";
import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status";
import { generateWeightStatusForUser } from "@/lib/insights/weight-status";
@@ -77,6 +79,16 @@ const QUEUE_NAME = "medication-reminder-check";
const CHECK_INTERVAL_CRON = "*/15 * * * *"; // every 15 minutes
const WITHINGS_SYNC_QUEUE = "withings-fallback-sync";
const WITHINGS_SYNC_CRON = "0 * * * *"; // every 60 minutes
// v1.4.25 W17b/c — webhook-primary + cron-safety-net for activity and
// sleep v2. The webhook handler enqueues per-user jobs on appli=16 / 44
// notifications; the crons below are the catch-net for the 1 % of
// notifications Withings drops. Offset 15-minute cadence per the
// research recommendation so the two queues don't lockstep against the
// existing measure cron at :00.
const WITHINGS_ACTIVITY_QUEUE = "withings-activity-sync";
const WITHINGS_ACTIVITY_CRON = "0 * * * *"; // every hour at :00
const WITHINGS_SLEEP_QUEUE = "withings-sleep-sync";
const WITHINGS_SLEEP_CRON = "15 * * * *"; // every hour at :15
const GENERAL_STATUS_QUEUE = "insights-general-status";
const GENERAL_STATUS_CRON = "0 2 * * *"; // daily at 02:00
const BLOOD_PRESSURE_STATUS_QUEUE = "insights-blood-pressure-status";
@@ -124,6 +136,26 @@ interface WithingsSyncPayload {
triggeredAt: string;
}
/**
* v1.4.25 W17b — payload for the activity-sync queue. When enqueued
* by the webhook handler, `userId` is set so the worker syncs only
* that user; when enqueued by the cron schedule, `userId` is absent
* and the worker iterates every connection (safety-net behaviour).
*/
interface WithingsActivitySyncPayload {
triggeredAt: string;
userId?: string;
}
/**
* v1.4.25 W17c — payload for the sleep-sync queue. Same shape and
* webhook-vs-cron semantics as the activity payload.
*/
interface WithingsSleepSyncPayload {
triggeredAt: string;
userId?: string;
}
interface GeneralStatusPayload {
triggeredAt: string;
}
@@ -550,6 +582,128 @@ async function handleWithingsFallbackSync(jobs: Job<WithingsSyncPayload>[]) {
});
}
/**
* v1.4.25 W17b — activity-sync handler.
*
* Two enqueue paths feed this queue:
*
* 1. Webhook (appli=16) — payload carries `userId`, the handler
* runs `syncUserActivity` for that one user.
* 2. Cron (`withings-activity-sync` at :00 every hour) — payload
* has no `userId`, the handler iterates every Withings
* connection and re-syncs each. Catches the 1 % of webhook
* deliveries Withings drops.
*
* Sync failures per-user are logged as warnings; the queue carries on
* so one user's parked-at-reauth state doesn't starve every other
* connection on the cron tick.
*/
async function handleWithingsActivitySync(
jobs: Job<WithingsActivitySyncPayload>[],
) {
await withBackgroundEvent("job.withings_activity_sync", async (evt) => {
const prisma = getWorkerPrisma();
try {
const targets: Array<{ userId: string }> = [];
for (const job of jobs) {
if (job.data?.userId) {
targets.push({ userId: job.data.userId });
}
}
// No user-specific enqueue → cron fallback iterating everyone.
if (targets.length === 0) {
const connections = await prisma.withingsConnection.findMany({
select: { userId: true },
});
targets.push(...connections);
}
if (targets.length === 0) return;
let usersSynced = 0;
let measurementsImported = 0;
for (const { userId } of targets) {
try {
const imported = await syncUserActivity(userId);
usersSynced++;
measurementsImported += imported;
} catch (err) {
evt.addWarning(
`Withings activity sync failed for user ${userId}: ${err}`,
);
}
}
evt.setBackground({
task_name: "job.withings_activity_sync",
result: {
users_synced: usersSynced,
total: targets.length,
measurements_imported: measurementsImported,
},
});
} catch (err) {
evt.setError(err);
recordError();
throw err;
}
});
}
/**
* v1.4.25 W17c — sleep-sync handler. Same enqueue semantics as the
* activity handler: per-user when the webhook fires, full-iteration
* when the cron ticks.
*/
async function handleWithingsSleepSync(
jobs: Job<WithingsSleepSyncPayload>[],
) {
await withBackgroundEvent("job.withings_sleep_sync", async (evt) => {
const prisma = getWorkerPrisma();
try {
const targets: Array<{ userId: string }> = [];
for (const job of jobs) {
if (job.data?.userId) {
targets.push({ userId: job.data.userId });
}
}
if (targets.length === 0) {
const connections = await prisma.withingsConnection.findMany({
select: { userId: true },
});
targets.push(...connections);
}
if (targets.length === 0) return;
let usersSynced = 0;
let measurementsImported = 0;
for (const { userId } of targets) {
try {
const imported = await syncUserSleep(userId);
usersSynced++;
measurementsImported += imported;
} catch (err) {
evt.addWarning(
`Withings sleep sync failed for user ${userId}: ${err}`,
);
}
}
evt.setBackground({
task_name: "job.withings_sleep_sync",
result: {
users_synced: usersSynced,
total: targets.length,
measurements_imported: measurementsImported,
},
});
} catch (err) {
evt.setError(err);
recordError();
throw err;
}
});
}
async function handleGeneralStatusGenerate(jobs: Job<GeneralStatusPayload>[]) {
void jobs;
await withBackgroundEvent("job.insights.general", async (evt) => {
@@ -1203,6 +1357,8 @@ export async function startReminderWorker() {
const allQueues = [
QUEUE_NAME,
WITHINGS_SYNC_QUEUE,
WITHINGS_ACTIVITY_QUEUE,
WITHINGS_SLEEP_QUEUE,
GENERAL_STATUS_QUEUE,
BLOOD_PRESSURE_STATUS_QUEUE,
WEIGHT_STATUS_QUEUE,
@@ -1228,6 +1384,8 @@ export async function startReminderWorker() {
const schedules: [string, string][] = [
[QUEUE_NAME, CHECK_INTERVAL_CRON],
[WITHINGS_SYNC_QUEUE, WITHINGS_SYNC_CRON],
[WITHINGS_ACTIVITY_QUEUE, WITHINGS_ACTIVITY_CRON],
[WITHINGS_SLEEP_QUEUE, WITHINGS_SLEEP_CRON],
[GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON],
[BLOOD_PRESSURE_STATUS_QUEUE, BLOOD_PRESSURE_STATUS_CRON],
[WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON],
@@ -1259,6 +1417,16 @@ export async function startReminderWorker() {
{ localConcurrency: 1 },
handleWithingsFallbackSync,
);
await boss.work<WithingsActivitySyncPayload>(
WITHINGS_ACTIVITY_QUEUE,
{ localConcurrency: 1 },
handleWithingsActivitySync,
);
await boss.work<WithingsSleepSyncPayload>(
WITHINGS_SLEEP_QUEUE,
{ localConcurrency: 1 },
handleWithingsSleepSync,
);
await boss.work<GeneralStatusPayload>(
GENERAL_STATUS_QUEUE,
{ localConcurrency: 1 },
+92 -6
View File
@@ -29,6 +29,19 @@ import { syncUserMeasurements } from "@/lib/withings/sync";
import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit";
import { getClientIp, safeJson } from "@/lib/api-response";
import { annotate, getEvent } from "@/lib/logging/context";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
/**
* v1.4.25 W17b/c — appli categories that have their own sync routine
* fan-out. The measure path (appli=1/2/4) keeps the legacy inline
* call to `syncUserMeasurements`; activity (16) and sleep v2 (44) hand
* off to pg-boss so the webhook response stays fast and a slow sync
* doesn't block subsequent deliveries on the same connection.
*/
const WITHINGS_ACTIVITY_APPLI = 16;
const WITHINGS_SLEEP_APPLI = 44;
const WITHINGS_ACTIVITY_QUEUE = "withings-activity-sync";
const WITHINGS_SLEEP_QUEUE = "withings-sleep-sync";
export type WithingsWebhookAuthOutcome =
| { ok: true }
@@ -84,23 +97,39 @@ export async function processWithingsNotification(
const contentType = request.headers.get("content-type") ?? "";
let withingsUserId: string | null = null;
let appli: number | null = null;
if (contentType.includes("application/x-www-form-urlencoded")) {
const formData = await request.formData();
withingsUserId = formData.get("userid") as string;
const rawAppli = formData.get("appli");
if (rawAppli != null) {
const parsed = Number.parseInt(String(rawAppli), 10);
if (Number.isFinite(parsed)) appli = parsed;
}
} else {
const { data: body, error: jsonError } = await safeJson<{
userid?: string | number;
appli?: string | number;
}>(request);
if (jsonError) return jsonError;
withingsUserId = body.userid?.toString() ?? null;
if (body.appli != null) {
const parsed = Number.parseInt(String(body.appli), 10);
if (Number.isFinite(parsed)) appli = parsed;
}
}
if (!withingsUserId) {
return NextResponse.json({ status: "ignored" }, { status: 200 });
}
annotate({ meta: { withings_user_id: withingsUserId } });
annotate({
meta: {
withings_user_id: withingsUserId,
withings_appli: appli ?? null,
},
});
const connection = await prisma.withingsConnection.findFirst({
where: { withingsUserId },
@@ -113,11 +142,68 @@ export async function processWithingsNotification(
return NextResponse.json({ status: "unknown_user" }, { status: 200 });
}
syncUserMeasurements(connection.userId).catch((err) => {
getEvent()?.addWarning(
"Sync failed for user " + connection.userId + ": " + err,
);
});
// v1.4.25 W17b/c — dispatch on appli category. Activity (16) and
// sleep v2 (44) hand off to dedicated pg-boss queues so the webhook
// response stays sub-100ms; the measure family (appli=1/2/4) keeps
// the existing inline `syncUserMeasurements` call. An absent or
// unknown `appli` defaults to the measure path — preserves the
// pre-W17 behaviour for legacy subscriptions Withings hasn't
// re-tagged yet.
if (appli === WITHINGS_ACTIVITY_APPLI) {
await enqueueWithingsSync(WITHINGS_ACTIVITY_QUEUE, connection.userId);
} else if (appli === WITHINGS_SLEEP_APPLI) {
await enqueueWithingsSync(WITHINGS_SLEEP_QUEUE, connection.userId);
} else {
syncUserMeasurements(connection.userId).catch((err) => {
getEvent()?.addWarning(
"Sync failed for user " + connection.userId + ": " + err,
);
});
}
return NextResponse.json({ status: "ok" }, { status: 200 });
}
/**
* v1.4.25 W17b/c — enqueue a per-user activity / sleep sync job onto
* the pg-boss queue. When pg-boss isn't available (Next.js dev server
* without the worker process attached) we fall back to firing the
* sync inline so dev iteration isn't blocked on a separate worker
* process — same behaviour as the legacy measure path.
*/
async function enqueueWithingsSync(
queueName: string,
userId: string,
): Promise<void> {
const boss = getGlobalBoss();
if (!boss) {
getEvent()?.addWarning(
`pg-boss not initialised — falling back to inline ${queueName} for ${userId}`,
);
// Lazy-load so the webhook route doesn't pull in the activity /
// sleep sync modules at cold-start when the worker process is up.
if (queueName === WITHINGS_ACTIVITY_QUEUE) {
const { syncUserActivity } = await import("./sync-activity");
syncUserActivity(userId).catch((err) => {
getEvent()?.addWarning(
`Inline activity sync failed for ${userId}: ${err}`,
);
});
} else if (queueName === WITHINGS_SLEEP_QUEUE) {
const { syncUserSleep } = await import("./sync-sleep");
syncUserSleep(userId).catch((err) => {
getEvent()?.addWarning(
`Inline sleep sync failed for ${userId}: ${err}`,
);
});
}
return;
}
try {
await boss.send(queueName, { userId, triggeredAt: new Date().toISOString() });
} catch (err) {
getEvent()?.addWarning(
`Failed to enqueue ${queueName} for ${userId}: ${err}`,
);
}
}