From b55942fd186517d10f315cdf69f7c168a2693ed6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 14 May 2026 18:29:59 +0200 Subject: [PATCH] feat(jobs): pg-boss queues for activity-sync + sleep-v2-sync + webhook enqueue hooks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../webhook/[token]/__tests__/route.test.ts | 13 ++ .../withings/webhook/__tests__/route.test.ts | 137 ++++++++++++++ .../jobs/__tests__/withings-queues.test.ts | 70 ++++++++ src/lib/jobs/reminder-worker.ts | 168 ++++++++++++++++++ src/lib/withings/webhook-handler.ts | 98 +++++++++- 5 files changed, 480 insertions(+), 6 deletions(-) create mode 100644 src/lib/jobs/__tests__/withings-queues.test.ts diff --git a/src/app/api/withings/webhook/[token]/__tests__/route.test.ts b/src/app/api/withings/webhook/[token]/__tests__/route.test.ts index 5e8251c39..480f34946 100644 --- a/src/app/api/withings/webhook/[token]/__tests__/route.test.ts +++ b/src/app/api/withings/webhook/[token]/__tests__/route.test.ts @@ -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(() => ({})), diff --git a/src/app/api/withings/webhook/__tests__/route.test.ts b/src/app/api/withings/webhook/__tests__/route.test.ts index a0c96e202..a273789cb 100644 --- a/src/app/api/withings/webhook/__tests__/route.test.ts +++ b/src/app/api/withings/webhook/__tests__/route.test.ts @@ -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", { diff --git a/src/lib/jobs/__tests__/withings-queues.test.ts b/src/lib/jobs/__tests__/withings-queues.test.ts new file mode 100644 index 000000000..8e4d27505 --- /dev/null +++ b/src/lib/jobs/__tests__/withings-queues.test.ts @@ -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\]/); + }); +}); diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index b84b0717c..0f111d277 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -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[]) { }); } +/** + * 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[], +) { + 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[], +) { + 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[]) { 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( + WITHINGS_ACTIVITY_QUEUE, + { localConcurrency: 1 }, + handleWithingsActivitySync, + ); + await boss.work( + WITHINGS_SLEEP_QUEUE, + { localConcurrency: 1 }, + handleWithingsSleepSync, + ); await boss.work( GENERAL_STATUS_QUEUE, { localConcurrency: 1 }, diff --git a/src/lib/withings/webhook-handler.ts b/src/lib/withings/webhook-handler.ts index 803badb10..16f645d69 100644 --- a/src/lib/withings/webhook-handler.ts +++ b/src/lib/withings/webhook-handler.ts @@ -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 { + 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}`, + ); + } +}