diff --git a/src/app/api/nightscout/connect/__tests__/route.test.ts b/src/app/api/nightscout/connect/__tests__/route.test.ts index 115bfb8a3..da7d9c03c 100644 --- a/src/app/api/nightscout/connect/__tests__/route.test.ts +++ b/src/app/api/nightscout/connect/__tests__/route.test.ts @@ -4,10 +4,14 @@ import { NextRequest } from "next/server"; import { SafeFetchError } from "@/lib/safe-fetch"; import { NightscoutApiError } from "@/lib/nightscout/client"; -const { fetchSgvEntriesMock, rateLimitMock } = vi.hoisted(() => ({ - fetchSgvEntriesMock: vi.fn(), - rateLimitMock: vi.fn(), -})); +const { fetchSgvEntriesMock, rateLimitMock, bossSend, bossMock } = vi.hoisted( + () => ({ + fetchSgvEntriesMock: vi.fn(), + rateLimitMock: vi.fn(), + bossSend: vi.fn(), + bossMock: vi.fn(), + }), +); vi.mock("@/lib/api-handler", () => ({ apiHandler: unknown>(fn: T) => fn, @@ -18,10 +22,14 @@ vi.mock("@/lib/db", () => ({ prisma: { user: { update: vi.fn() } }, })); -vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); +vi.mock("@/lib/logging/context", () => ({ + annotate: vi.fn(), + getEvent: () => ({ addWarning: vi.fn() }), +})); vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() })); vi.mock("@/lib/crypto", () => ({ encrypt: (s: string) => `enc:${s}` })); vi.mock("@/lib/integrations/status", () => ({ markReconnected: vi.fn() })); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: bossMock })); vi.mock("@/lib/nightscout/client", async (importOriginal) => { const actual = await importOriginal(); @@ -66,6 +74,8 @@ beforeEach(() => { vi.clearAllMocks(); rateLimitMock.mockResolvedValue({ allowed: true, remaining: 9, resetAt: 0 }); fetchSgvEntriesMock.mockResolvedValue([{ id: "x", sgv: 100, date: 1 }]); + bossSend.mockResolvedValue("job-1"); + bossMock.mockReturnValue({ send: bossSend }); }); describe("POST /api/nightscout/connect", () => { @@ -208,3 +218,47 @@ describe("POST /api/nightscout/connect", () => { expect(fetchSgvEntriesMock).not.toHaveBeenCalled(); }); }); + +/** + * v1.32.28 — a Nightscout connection pulls its first glucose window straight + * away instead of leaving the card empty until the next hourly tick. + */ +describe("POST /api/nightscout/connect — connect-time first sync", () => { + it("enqueues exactly one targeted nightscout-sync job for the connecting user", async () => { + const res = await post( + req({ url: "https://ns.example.com", token: "tok" }), + ); + + expect(res.status).toBe(200); + expect(bossSend).toHaveBeenCalledTimes(1); + expect(bossSend).toHaveBeenCalledWith("nightscout-sync", { userId: "u1" }); + }); + + it("does not enqueue when the connection was rejected", async () => { + const res = await post(req({ url: "not a url" })); + + expect(res.status).toBe(422); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("still connects when no boss instance is available", async () => { + bossMock.mockReturnValue(null); + + const res = await post( + req({ url: "https://ns.example.com", token: "tok" }), + ); + + expect(res.status).toBe(200); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("still connects when the enqueue itself rejects", async () => { + bossSend.mockRejectedValue(new Error("queue unreachable")); + + const res = await post( + req({ url: "https://ns.example.com", token: "tok" }), + ); + + expect(res.status).toBe(200); + }); +}); diff --git a/src/app/api/nightscout/connect/route.ts b/src/app/api/nightscout/connect/route.ts index 50c7ad821..6108e05cc 100644 --- a/src/app/api/nightscout/connect/route.ts +++ b/src/app/api/nightscout/connect/route.ts @@ -4,11 +4,13 @@ import { z } from "zod/v4"; import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { apiSuccess, apiError, safeJson } from "@/lib/api-response"; -import { annotate } from "@/lib/logging/context"; +import { annotate, getEvent } from "@/lib/logging/context"; import { auditLog } from "@/lib/auth/audit"; import { encrypt } from "@/lib/crypto"; import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; import { markReconnected } from "@/lib/integrations/status"; +import { NIGHTSCOUT_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { SafeFetchError } from "@/lib/safe-fetch"; import { fetchSgvEntries, NightscoutApiError } from "@/lib/nightscout/client"; import { nightscoutConnectSchema } from "@/lib/validations/nightscout"; @@ -116,5 +118,20 @@ export const POST = apiHandler(async (request: NextRequest) => { await markReconnected(user.id, "nightscout"); await auditLog("nightscout.connect", { userId: user.id }); + // Pull the first glucose window now instead of leaving the user staring at an + // empty card until the next hourly tick. Best-effort by design: the hourly + // cron remains the safety net, so a missing boss instance must never fail the + // connect response the user just earned. + const boss = getGlobalBoss(); + if (boss) { + await boss + .send(NIGHTSCOUT_SYNC_QUEUE, { userId: user.id }) + .catch((err) => + getEvent()?.addWarning( + `nightscout-sync connect enqueue failed: ${err}`, + ), + ); + } + return apiSuccess({ connected: true }); }); diff --git a/src/app/api/nightscout/sync/__tests__/route.test.ts b/src/app/api/nightscout/sync/__tests__/route.test.ts new file mode 100644 index 000000000..a133cb8c5 --- /dev/null +++ b/src/app/api/nightscout/sync/__tests__/route.test.ts @@ -0,0 +1,116 @@ +/** + * v1.32.28 — manual Nightscout sync. + * + * The error arm carries the load here: a Nightscout base URL carries the API + * token as a query parameter, so an upstream message can contain the user's + * secret. The route answers with a fixed string and never the caught message. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; +import type * as ApiHandlerModule from "@/lib/api-handler"; + +vi.mock("@/lib/api-handler", async () => { + const actual = + await vi.importActual("@/lib/api-handler"); + return { + ...actual, + requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), + }; +}); + +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); + +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(async () => ({ allowed: true })), +})); + +vi.mock("@/lib/nightscout/sync", () => ({ syncUserNightscout: vi.fn() })); + +import { POST } from "../route"; +import { HttpError, requireAuth } from "@/lib/api-handler"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { syncUserNightscout } from "@/lib/nightscout/sync"; + +function request(): NextRequest { + return new NextRequest("http://localhost/api/nightscout/sync", { + method: "POST", + }); +} + +// The route takes no arguments (no body, no flags); the alias keeps the +// direct invoke typed while still handing apiHandler a real request. +const post = POST as unknown as (r: NextRequest) => Promise; + +async function envelope(response: Response) { + return (await response.json()) as { + data: { imported?: number } | null; + error: string | null; + meta?: { errorCode?: string }; + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: true } as never); + vi.mocked(requireAuth).mockResolvedValue({ user: { id: "u1" } } as never); +}); + +describe("POST /api/nightscout/sync", () => { + it("syncs the session user and answers with the freshly inserted count", async () => { + vi.mocked(syncUserNightscout).mockResolvedValue(36); + + const response = await post(request()); + + expect(response.status).toBe(200); + expect(await envelope(response)).toEqual({ + data: { imported: 36 }, + error: null, + }); + expect(syncUserNightscout).toHaveBeenCalledWith("u1"); + }); + + it("refuses an unauthenticated caller before syncing anything", async () => { + vi.mocked(requireAuth).mockRejectedValue( + new HttpError(401, "Not authenticated"), + ); + + const response = await post(request()); + + expect(response.status).toBe(401); + expect(syncUserNightscout).not.toHaveBeenCalled(); + }); + + it("refuses past the per-user rate limit without touching the sync", async () => { + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: false } as never); + + const response = await post(request()); + const body = await envelope(response); + + expect(response.status).toBe(429); + expect(body.meta?.errorCode).toBe("rate_limited_self"); + expect(syncUserNightscout).not.toHaveBeenCalled(); + expect(checkRateLimit).toHaveBeenCalledWith( + "nightscout-sync:u1", + 5, + 60_000, + ); + }); + + it("never echoes the token-bearing instance URL into the response body", async () => { + vi.mocked(syncUserNightscout).mockRejectedValue( + new Error( + "GET https://cgm.example.org/api/v1/entries.json?token=hunter2secret failed with 401", + ), + ); + + const response = await post(request()); + const body = await envelope(response); + const serialised = JSON.stringify(body); + + expect(response.status).toBe(502); + expect(body.error).toBe("Nightscout sync failed"); + expect(serialised).not.toContain("hunter2secret"); + expect(serialised).not.toContain("cgm.example.org"); + expect(serialised).not.toContain("token="); + }); +}); diff --git a/src/app/api/nightscout/sync/route.ts b/src/app/api/nightscout/sync/route.ts new file mode 100644 index 000000000..1c379ce79 --- /dev/null +++ b/src/app/api/nightscout/sync/route.ts @@ -0,0 +1,42 @@ +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { apiSuccess, apiError } from "@/lib/api-response"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { annotate } from "@/lib/logging/context"; +import { syncUserNightscout } from "@/lib/nightscout/sync"; + +/** + * Manually trigger a Nightscout sync for the current user (v1.32.28). + * + * Pulls the recent SGV window from the configured instance; `imported` counts + * the readings that were genuinely new this run, not the ones re-confirmed. + * + * No body and no `fullSync` flag: there is no full-history arm to call. + * + * The error arm answers with a fixed message and NEVER the caught error's own. + * A Nightscout base URL carries the API token as a query parameter, so an + * upstream message can contain the user's secret; echoing it here would put it + * in a response body, a browser devtools panel, and any proxy log in between. + * The classified failure is already on the `nightscout` ledger, recorded by the + * provider with its redaction rules applied. + * + * Rate-limited 5/60s per user, matching the hardened Fitbit / Google Health + * routes; every run is an outbound call to a user-supplied host. + */ +export const POST = apiHandler(async () => { + const { user } = await requireAuth(); + annotate({ action: { name: "nightscout.sync" } }); + + const rl = await checkRateLimit(`nightscout-sync:${user.id}`, 5, 60_000); + if (!rl.allowed) { + return apiError("Too many sync requests", 429, { + errorCode: "rate_limited_self", + }); + } + + try { + const imported = await syncUserNightscout(user.id); + return apiSuccess({ imported }); + } catch { + return apiError("Nightscout sync failed", 502); + } +}); diff --git a/src/app/api/oura/callback/__tests__/route.test.ts b/src/app/api/oura/callback/__tests__/route.test.ts index acd65851c..07cb60d32 100644 --- a/src/app/api/oura/callback/__tests__/route.test.ts +++ b/src/app/api/oura/callback/__tests__/route.test.ts @@ -12,21 +12,34 @@ vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn(async () => null) })); vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn(), - getEvent: () => ({ setError: vi.fn(), setAuth: vi.fn() }), + getEvent: () => ({ + setError: vi.fn(), + setAuth: vi.fn(), + addWarning: vi.fn(), + }), })); vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() })); vi.mock("@/lib/crypto", () => ({ encrypt: (s: string) => `enc:${s}` })); vi.mock("@/lib/integrations/status", () => ({ markReconnected: vi.fn() })); -const { exchangeMock, getCredsMock, matchMock, verifyMock } = vi.hoisted( - () => ({ - exchangeMock: vi.fn(), - getCredsMock: vi.fn(), - matchMock: vi.fn(), - verifyMock: vi.fn(), - }), -); +const { + exchangeMock, + getCredsMock, + matchMock, + verifyMock, + bossSend, + bossMock, +} = vi.hoisted(() => ({ + exchangeMock: vi.fn(), + getCredsMock: vi.fn(), + matchMock: vi.fn(), + verifyMock: vi.fn(), + bossSend: vi.fn(), + bossMock: vi.fn(), +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: bossMock })); vi.mock("@/lib/oura/client", () => ({ exchangeCode: exchangeMock })); vi.mock("@/lib/oura/credentials", () => ({ @@ -75,6 +88,8 @@ beforeEach(() => { exchangeMock.mockResolvedValue({ access_token: "at", refresh_token: "rt" }); userUpdate.mockResolvedValue({}); (getSession as ReturnType).mockResolvedValue(null); + bossSend.mockResolvedValue("job-1"); + bossMock.mockReturnValue({ send: bossSend }); }); describe("GET /api/oura/callback", () => { @@ -123,3 +138,36 @@ describe("GET /api/oura/callback", () => { expect(res.headers.get("location")).toContain("oura=connected"); }); }); + +/** + * v1.32.28 — connecting starts a sync straight away. The poll handler has + * accepted a single-user payload since it was written; this callback is the + * producer that was missing, and without it a fresh connection showed an empty + * card for up to an hour. + */ +describe("GET /api/oura/callback — connect-time first sync", () => { + it("enqueues exactly one targeted oura-sync job for the connecting user", async () => { + const res = await GET(makeReq({ code: "c", state: "S", cookie: "S" })); + + expect(res.headers.get("location")).toContain("oura=connected"); + expect(bossSend).toHaveBeenCalledTimes(1); + expect(bossSend).toHaveBeenCalledWith("oura-sync", { userId: "u1" }); + }); + + it("still connects when no boss instance is available", async () => { + bossMock.mockReturnValue(null); + + const res = await GET(makeReq({ code: "c", state: "S", cookie: "S" })); + + expect(res.headers.get("location")).toContain("oura=connected"); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("still connects when the enqueue itself rejects", async () => { + bossSend.mockRejectedValue(new Error("queue unreachable")); + + const res = await GET(makeReq({ code: "c", state: "S", cookie: "S" })); + + expect(res.headers.get("location")).toContain("oura=connected"); + }); +}); diff --git a/src/app/api/oura/callback/route.ts b/src/app/api/oura/callback/route.ts index 9dd9a0fad..553dd9c64 100644 --- a/src/app/api/oura/callback/route.ts +++ b/src/app/api/oura/callback/route.ts @@ -12,6 +12,8 @@ import { verifySignedState, } from "@/lib/oauth/signed-state"; import { markReconnected } from "@/lib/integrations/status"; +import { OURA_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { NextRequest, NextResponse } from "next/server"; /** @@ -82,6 +84,21 @@ export const GET = apiHandler(async (request: NextRequest) => { await auditLog("oura.connect", { userId }); await markReconnected(userId, "oura"); + // Start the first sync now instead of leaving the user staring at an empty + // card until the next hourly tick. The poll handler has accepted a + // single-user payload since it was written; this is the producer that was + // missing. Reconnects ride the same path. Best-effort by design: the hourly + // cron remains the safety net, so a missing boss instance must never strand + // the redirect. + const boss = getGlobalBoss(); + if (boss) { + await boss + .send(OURA_SYNC_QUEUE, { userId }) + .catch((err) => + getEvent()?.addWarning(`oura-sync connect enqueue failed: ${err}`), + ); + } + const response = NextResponse.redirect( new URL( "/settings/integrations?oura=connected", diff --git a/src/app/api/oura/sync/__tests__/route.test.ts b/src/app/api/oura/sync/__tests__/route.test.ts new file mode 100644 index 000000000..8dd1fd51a --- /dev/null +++ b/src/app/api/oura/sync/__tests__/route.test.ts @@ -0,0 +1,104 @@ +/** + * v1.32.28 — manual Oura sync. Until now Oura only ran on its hourly tick. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; +import type * as ApiHandlerModule from "@/lib/api-handler"; + +vi.mock("@/lib/api-handler", async () => { + const actual = + await vi.importActual("@/lib/api-handler"); + return { + ...actual, + requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), + }; +}); + +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); + +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(async () => ({ allowed: true })), +})); + +vi.mock("@/lib/oura/sync", () => ({ syncUserOura: vi.fn() })); + +import { POST } from "../route"; +import { HttpError, requireAuth } from "@/lib/api-handler"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { syncUserOura } from "@/lib/oura/sync"; + +function request(): NextRequest { + return new NextRequest("http://localhost/api/oura/sync", { method: "POST" }); +} + +// The route takes no arguments (no body, no flags); the alias keeps the +// direct invoke typed while still handing apiHandler a real request. +const post = POST as unknown as (r: NextRequest) => Promise; + +async function envelope(response: Response) { + return (await response.json()) as { + data: { imported?: number } | null; + error: string | null; + meta?: { errorCode?: string }; + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: true } as never); + vi.mocked(requireAuth).mockResolvedValue({ + user: { id: "u1" }, + } as never); +}); + +describe("POST /api/oura/sync", () => { + it("syncs the session user and answers with the imported count", async () => { + vi.mocked(syncUserOura).mockResolvedValue(12); + + const response = await post(request()); + + expect(response.status).toBe(200); + expect(await envelope(response)).toEqual({ + data: { imported: 12 }, + error: null, + }); + // The id comes from the session, never from the request. + expect(syncUserOura).toHaveBeenCalledWith("u1"); + }); + + it("refuses an unauthenticated caller before syncing anything", async () => { + vi.mocked(requireAuth).mockRejectedValue( + new HttpError(401, "Not authenticated"), + ); + + const response = await post(request()); + + expect(response.status).toBe(401); + expect(syncUserOura).not.toHaveBeenCalled(); + }); + + it("refuses past the per-user rate limit without touching the sync", async () => { + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: false } as never); + + const response = await post(request()); + const body = await envelope(response); + + expect(response.status).toBe(429); + expect(body.meta?.errorCode).toBe("rate_limited_self"); + expect(syncUserOura).not.toHaveBeenCalled(); + expect(checkRateLimit).toHaveBeenCalledWith("oura-sync:u1", 5, 60_000); + }); + + it("answers 502 generically when the sync throws", async () => { + vi.mocked(syncUserOura).mockRejectedValue( + new Error("oura said: token abc123 is invalid"), + ); + + const response = await post(request()); + const body = await envelope(response); + + expect(response.status).toBe(502); + expect(body.error).toBe("Oura sync failed"); + expect(JSON.stringify(body)).not.toContain("abc123"); + }); +}); diff --git a/src/app/api/oura/sync/route.ts b/src/app/api/oura/sync/route.ts new file mode 100644 index 000000000..98602ed80 --- /dev/null +++ b/src/app/api/oura/sync/route.ts @@ -0,0 +1,42 @@ +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { apiSuccess, apiError } from "@/lib/api-response"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { annotate } from "@/lib/logging/context"; +import { syncUserOura } from "@/lib/oura/sync"; + +/** + * Manually trigger an Oura sync for the current user (v1.32.28). + * + * Until now Oura only ever synced on its hourly cron tick, so a user who could + * see something was behind had no way to ask for a fresh pull. + * + * No body and no `fullSync` flag: `syncUserOura` has no full-history arm, and + * the window is not the caller's to choose — the sync derives it from the + * newest row it already holds, which is what makes a catch-up honest. Accepting + * a `lookbackDays` here would hand an unvalidated cost knob to the client. + * + * Rate-limited 5/60s per user, matching the hardened Fitbit / Google Health + * routes; a sync is a fan-out of upstream requests plus a batched write, so a + * tight retry loop must not be able to pin Prisma or burn the Oura budget. + */ +export const POST = apiHandler(async () => { + const { user } = await requireAuth(); + annotate({ action: { name: "oura.sync" } }); + + const rl = await checkRateLimit(`oura-sync:${user.id}`, 5, 60_000); + if (!rl.allowed) { + return apiError("Too many sync requests", 429, { + errorCode: "rate_limited_self", + }); + } + + try { + const imported = await syncUserOura(user.id); + return apiSuccess({ imported }); + } catch { + // The failure is already on the `oura` ledger, classified by the provider + // that owns the error shape. Answer generically — an upstream message can + // carry detail that has no business in a response body. + return apiError("Oura sync failed", 502); + } +}); diff --git a/src/app/api/polar/callback/__tests__/route.test.ts b/src/app/api/polar/callback/__tests__/route.test.ts index c290363ad..ac69bba3e 100644 --- a/src/app/api/polar/callback/__tests__/route.test.ts +++ b/src/app/api/polar/callback/__tests__/route.test.ts @@ -8,6 +8,8 @@ const { registerUserMock, getCredsMock, markReconnectedMock, + bossSend, + bossMock, } = vi.hoisted(() => ({ stateMatchesCookieMock: vi.fn(), verifySignedStateMock: vi.fn(), @@ -16,6 +18,8 @@ const { registerUserMock: vi.fn(), getCredsMock: vi.fn(), markReconnectedMock: vi.fn(), + bossSend: vi.fn(), + bossMock: vi.fn(), })); vi.mock("@/lib/api-handler", () => ({ @@ -24,7 +28,11 @@ vi.mock("@/lib/api-handler", () => ({ vi.mock("@/lib/db", () => ({ prisma: { user: { update: vi.fn() } } })); vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn(), - getEvent: () => ({ setAuth: vi.fn(), setError: vi.fn() }), + getEvent: () => ({ + setAuth: vi.fn(), + setError: vi.fn(), + addWarning: vi.fn(), + }), })); vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() })); vi.mock("@/lib/crypto", () => ({ encrypt: (s: string) => `enc:${s}` })); @@ -44,6 +52,7 @@ vi.mock("@/lib/oauth/signed-state", () => ({ vi.mock("@/lib/integrations/status", () => ({ markReconnected: markReconnectedMock, })); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: bossMock })); import { GET } from "../route"; @@ -68,6 +77,8 @@ beforeEach(() => { getSessionMock.mockResolvedValue(null); getCredsMock.mockResolvedValue({ clientId: "c", clientSecret: "s" }); exchangeCodeMock.mockResolvedValue({ access_token: "tok", x_user_id: 42 }); + bossSend.mockResolvedValue("job-1"); + bossMock.mockReturnValue({ send: bossSend }); registerUserMock.mockResolvedValue(undefined); }); @@ -100,3 +111,35 @@ describe("GET /api/polar/callback", () => { expect(registerUserMock).not.toHaveBeenCalled(); }); }); + +/** + * v1.32.28 — connecting starts a sync straight away instead of leaving the card + * empty until the next hourly tick. The poll handler has accepted a single-user + * payload since it was written; this callback is the producer that was missing. + */ +describe("GET /api/polar/callback — connect-time first sync", () => { + it("enqueues exactly one targeted polar-sync job for the connecting user", async () => { + const res = await run(makeReq()); + + expect(res.headers.get("location")).toContain("polar=connected"); + expect(bossSend).toHaveBeenCalledTimes(1); + expect(bossSend).toHaveBeenCalledWith("polar-sync", { userId: "u1" }); + }); + + it("still connects when no boss instance is available", async () => { + bossMock.mockReturnValue(null); + + const res = await run(makeReq()); + + expect(res.headers.get("location")).toContain("polar=connected"); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("still connects when the enqueue itself rejects", async () => { + bossSend.mockRejectedValue(new Error("queue unreachable")); + + const res = await run(makeReq()); + + expect(res.headers.get("location")).toContain("polar=connected"); + }); +}); diff --git a/src/app/api/polar/callback/route.ts b/src/app/api/polar/callback/route.ts index e8a8a09f7..8e89f51b1 100644 --- a/src/app/api/polar/callback/route.ts +++ b/src/app/api/polar/callback/route.ts @@ -12,6 +12,8 @@ import { verifySignedState, } from "@/lib/oauth/signed-state"; import { markReconnected } from "@/lib/integrations/status"; +import { POLAR_SYNC_QUEUE } from "@/lib/jobs/integration-poll-queues"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { NextRequest, NextResponse } from "next/server"; /** @@ -102,6 +104,19 @@ export const GET = apiHandler(async (request: NextRequest) => { await auditLog("polar.connect", { userId }); await markReconnected(userId, "polar"); + // Start the first sync now instead of leaving the user staring at an empty + // card until the next hourly tick. Best-effort by design: the hourly cron + // remains the safety net, so a missing boss instance must never strand the + // redirect. + const boss = getGlobalBoss(); + if (boss) { + await boss + .send(POLAR_SYNC_QUEUE, { userId }) + .catch((err) => + getEvent()?.addWarning(`polar-sync connect enqueue failed: ${err}`), + ); + } + const response = NextResponse.redirect( new URL( "/settings/integrations?polar=connected", diff --git a/src/app/api/polar/sync/__tests__/route.test.ts b/src/app/api/polar/sync/__tests__/route.test.ts new file mode 100644 index 000000000..f18d71c4f --- /dev/null +++ b/src/app/api/polar/sync/__tests__/route.test.ts @@ -0,0 +1,132 @@ +/** + * v1.32.28 — manual Polar sync. + * + * The entry point matters as much as the behaviour: the route must call + * `syncUserPolarLegs`, the same seam the hourly poll uses, so both legs run + * with the attribution rules built for them. Calling the vitals leg alone would + * quietly import half of Polar and stamp the ledger wrong, and a naive test + * that only asserted a row count would not notice. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { NextRequest } from "next/server"; +import type * as ApiHandlerModule from "@/lib/api-handler"; + +const { syncUserPolarMock, syncUserPolarWorkoutsMock } = vi.hoisted(() => ({ + syncUserPolarMock: vi.fn(), + syncUserPolarWorkoutsMock: vi.fn(), +})); + +vi.mock("@/lib/api-handler", async () => { + const actual = + await vi.importActual("@/lib/api-handler"); + return { + ...actual, + requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), + }; +}); + +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); + +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(async () => ({ allowed: true })), +})); + +// Mock the two LEGS rather than the wrapper, so the real `syncUserPolarLegs` +// runs and the test can see whether both of them were driven. +vi.mock("@/lib/polar/sync", () => ({ + syncUserPolar: syncUserPolarMock, + recordPolarSyncFailure: vi.fn(), +})); +vi.mock("@/lib/polar/sync-workouts", () => ({ + syncUserPolarWorkouts: syncUserPolarWorkoutsMock, +})); + +import { POST } from "../route"; +import { HttpError, requireAuth } from "@/lib/api-handler"; +import { checkRateLimit } from "@/lib/rate-limit"; + +function request(): NextRequest { + return new NextRequest("http://localhost/api/polar/sync", { method: "POST" }); +} + +// The route takes no arguments (no body, no flags); the alias keeps the +// direct invoke typed while still handing apiHandler a real request. +const post = POST as unknown as (r: NextRequest) => Promise; + +async function envelope(response: Response) { + return (await response.json()) as { + data: { imported?: number } | null; + error: string | null; + meta?: { errorCode?: string }; + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: true } as never); + vi.mocked(requireAuth).mockResolvedValue({ user: { id: "u1" } } as never); + syncUserPolarMock.mockResolvedValue(0); + syncUserPolarWorkoutsMock.mockResolvedValue(0); +}); + +describe("POST /api/polar/sync", () => { + it("runs both Polar legs and totals what they imported", async () => { + syncUserPolarMock.mockResolvedValue(5); + syncUserPolarWorkoutsMock.mockResolvedValue(2); + + const response = await post(request()); + + expect(response.status).toBe(200); + expect(await envelope(response)).toEqual({ + data: { imported: 7 }, + error: null, + }); + expect(syncUserPolarMock).toHaveBeenCalledWith("u1"); + expect(syncUserPolarWorkoutsMock).toHaveBeenCalledWith("u1"); + }); + + it("goes through the shared legs entry point, not the provider internals", () => { + const source = readFileSync(join(__dirname, "..", "route.ts"), "utf8"); + expect(source).toContain("syncUserPolarLegs"); + expect(source).not.toMatch(/from "@\/lib\/polar\//); + }); + + it("refuses an unauthenticated caller before syncing anything", async () => { + vi.mocked(requireAuth).mockRejectedValue( + new HttpError(401, "Not authenticated"), + ); + + const response = await post(request()); + + expect(response.status).toBe(401); + expect(syncUserPolarMock).not.toHaveBeenCalled(); + expect(syncUserPolarWorkoutsMock).not.toHaveBeenCalled(); + }); + + it("refuses past the per-user rate limit without touching the sync", async () => { + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: false } as never); + + const response = await post(request()); + const body = await envelope(response); + + expect(response.status).toBe(429); + expect(body.meta?.errorCode).toBe("rate_limited_self"); + expect(syncUserPolarMock).not.toHaveBeenCalled(); + expect(checkRateLimit).toHaveBeenCalledWith("polar-sync:u1", 5, 60_000); + }); + + it("answers 502 generically when a leg throws", async () => { + syncUserPolarMock.mockRejectedValue( + new Error("polar said: token abc123 is invalid"), + ); + + const response = await post(request()); + const body = await envelope(response); + + expect(response.status).toBe(502); + expect(body.error).toBe("Polar sync failed"); + expect(JSON.stringify(body)).not.toContain("abc123"); + }); +}); diff --git a/src/app/api/polar/sync/route.ts b/src/app/api/polar/sync/route.ts new file mode 100644 index 000000000..01b528dcd --- /dev/null +++ b/src/app/api/polar/sync/route.ts @@ -0,0 +1,41 @@ +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { apiSuccess, apiError } from "@/lib/api-response"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { annotate } from "@/lib/logging/context"; +import { syncUserPolarLegs } from "@/lib/jobs/reminder/polar-sync"; + +/** + * Manually trigger a Polar sync for the current user (v1.32.28). + * + * Calls `syncUserPolarLegs` — the same entry point the hourly poll uses — so + * both Polar legs run with the attribution rules that were built for them + * (vitals then workouts, independent failure recording, one success stamp). + * Reaching past it into `@/lib/polar/*` would run half the provider and get the + * ledger wrong. The module graph behind this import bottoms out in the shared + * Prisma client and constants, so it is safe on the request path. + * + * No body and no `fullSync` flag: there is no full-history arm to call. + * + * Rate-limited 5/60s per user, matching the hardened Fitbit / Google Health + * routes. + */ +export const POST = apiHandler(async () => { + const { user } = await requireAuth(); + annotate({ action: { name: "polar.sync" } }); + + const rl = await checkRateLimit(`polar-sync:${user.id}`, 5, 60_000); + if (!rl.allowed) { + return apiError("Too many sync requests", 429, { + errorCode: "rate_limited_self", + }); + } + + try { + const imported = await syncUserPolarLegs(user.id); + return apiSuccess({ imported }); + } catch { + // Both legs record their own classified failure on the `polar` ledger. + // Answer generically so no upstream detail rides the response body. + return apiError("Polar sync failed", 502); + } +}); diff --git a/src/app/api/strava/sync/__tests__/route.test.ts b/src/app/api/strava/sync/__tests__/route.test.ts new file mode 100644 index 000000000..45fcd1bcf --- /dev/null +++ b/src/app/api/strava/sync/__tests__/route.test.ts @@ -0,0 +1,104 @@ +/** + * v1.32.28 — manual Strava sync. Incremental by nature (Strava carries a real + * cursor); no full-history arm is exposed here, that stays with the backfill. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; +import type * as ApiHandlerModule from "@/lib/api-handler"; + +vi.mock("@/lib/api-handler", async () => { + const actual = + await vi.importActual("@/lib/api-handler"); + return { + ...actual, + requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), + }; +}); + +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); + +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(async () => ({ allowed: true })), +})); + +vi.mock("@/lib/strava/sync", () => ({ syncUserStrava: vi.fn() })); + +import { POST } from "../route"; +import { HttpError, requireAuth } from "@/lib/api-handler"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { syncUserStrava } from "@/lib/strava/sync"; + +function request(): NextRequest { + return new NextRequest("http://localhost/api/strava/sync", { + method: "POST", + }); +} + +// The route takes no arguments (no body, no flags); the alias keeps the +// direct invoke typed while still handing apiHandler a real request. +const post = POST as unknown as (r: NextRequest) => Promise; + +async function envelope(response: Response) { + return (await response.json()) as { + data: { imported?: number } | null; + error: string | null; + meta?: { errorCode?: string }; + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: true } as never); + vi.mocked(requireAuth).mockResolvedValue({ user: { id: "u1" } } as never); +}); + +describe("POST /api/strava/sync", () => { + it("syncs the session user and answers with the imported count", async () => { + vi.mocked(syncUserStrava).mockResolvedValue(4); + + const response = await post(request()); + + expect(response.status).toBe(200); + expect(await envelope(response)).toEqual({ + data: { imported: 4 }, + error: null, + }); + expect(syncUserStrava).toHaveBeenCalledWith("u1"); + }); + + it("refuses an unauthenticated caller before syncing anything", async () => { + vi.mocked(requireAuth).mockRejectedValue( + new HttpError(401, "Not authenticated"), + ); + + const response = await post(request()); + + expect(response.status).toBe(401); + expect(syncUserStrava).not.toHaveBeenCalled(); + }); + + it("refuses past the per-user rate limit without touching the sync", async () => { + vi.mocked(checkRateLimit).mockResolvedValue({ allowed: false } as never); + + const response = await post(request()); + const body = await envelope(response); + + expect(response.status).toBe(429); + expect(body.meta?.errorCode).toBe("rate_limited_self"); + expect(syncUserStrava).not.toHaveBeenCalled(); + expect(checkRateLimit).toHaveBeenCalledWith("strava-sync:u1", 5, 60_000); + }); + + it("answers 502 generically when the sync throws", async () => { + vi.mocked(syncUserStrava).mockRejectedValue( + new Error("strava said: token abc123 is invalid"), + ); + + const response = await post(request()); + const body = await envelope(response); + + expect(response.status).toBe(502); + expect(body.error).toBe("Strava sync failed"); + expect(JSON.stringify(body)).not.toContain("abc123"); + }); +}); diff --git a/src/app/api/strava/sync/route.ts b/src/app/api/strava/sync/route.ts new file mode 100644 index 000000000..0c36326cd --- /dev/null +++ b/src/app/api/strava/sync/route.ts @@ -0,0 +1,37 @@ +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { apiSuccess, apiError } from "@/lib/api-response"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { annotate } from "@/lib/logging/context"; +import { syncUserStrava } from "@/lib/strava/sync"; + +/** + * Manually trigger a Strava sync for the current user (v1.32.28). + * + * Incremental by nature: Strava already carries a real cursor + * (`stravaLastActivityAt` minus an overlap), so a manual run picks up wherever + * the last one stopped. No `fullSync` flag — deep history belongs to the + * connect-time backfill queue, which is self-converging and already gated. + * + * Rate-limited 5/60s per user, matching the hardened Fitbit / Google Health + * routes; Strava's 200-req / 15-min cap leaves no room for a retry loop. + */ +export const POST = apiHandler(async () => { + const { user } = await requireAuth(); + annotate({ action: { name: "strava.sync" } }); + + const rl = await checkRateLimit(`strava-sync:${user.id}`, 5, 60_000); + if (!rl.allowed) { + return apiError("Too many sync requests", 429, { + errorCode: "rate_limited_self", + }); + } + + try { + const imported = await syncUserStrava(user.id); + return apiSuccess({ imported }); + } catch { + // Already recorded on the `strava` ledger with its classification. Answer + // generically so no upstream detail rides the response body. + return apiError("Strava sync failed", 502); + } +}); diff --git a/src/lib/jobs/__tests__/nightscout-queues.test.ts b/src/lib/jobs/__tests__/nightscout-queues.test.ts index 373f019df..5873b8382 100644 --- a/src/lib/jobs/__tests__/nightscout-queues.test.ts +++ b/src/lib/jobs/__tests__/nightscout-queues.test.ts @@ -18,7 +18,15 @@ import { describe, expect, it } from "vitest"; // file into domain registrars; the Nightscout / Polar / Oura queues live in the // integration-sync registrar. The dead-queue guard follows the wiring there // (the handler module is still concatenated for the handler-symbol assertions). +// +// v1.32.28 — the four poll-queue NAMES moved again, into +// `integration-poll-queues.ts`, so an OAuth callback can enqueue a connect-time +// first sync without importing the registrar's handler graph. The guard follows +// them: the name declarations are asserted against that module, everything else +// (allQueues membership, cron schedule, handler binding) still against the +// registrar. const source = + readFileSync(join(__dirname, "..", "integration-poll-queues.ts"), "utf8") + readFileSync( join(__dirname, "..", "reminder", "register-integration-sync.ts"), "utf8", diff --git a/src/lib/jobs/integration-poll-queues.ts b/src/lib/jobs/integration-poll-queues.ts new file mode 100644 index 000000000..addd5e94d --- /dev/null +++ b/src/lib/jobs/integration-poll-queues.ts @@ -0,0 +1,27 @@ +/** + * Queue names for the poll-only integration syncs. + * + * Split out of `reminder/register-integration-sync.ts` so a request-path caller + * (an OAuth callback enqueueing the first sync at connect time) can name the + * queue without importing the registrar — which would drag every worker handler + * and its transitive graph into the request bundle. Same shape as + * `STRAVA_BACKFILL_QUEUE`, which the Strava callback already imports from its + * own module. + * + * The registrar remains the single place that creates, schedules, and binds a + * handler to each of these; the values here are its source. A name declared + * here but missing from `allQueues` there is the v1.4.37 dead-queue class, and + * the queue-wiring guards read both files for exactly that reason. + */ + +/** Hourly Nightscout CGM poll (:11). */ +export const NIGHTSCOUT_SYNC_QUEUE = "nightscout-sync"; + +/** Hourly Polar AccessLink poll (:13). */ +export const POLAR_SYNC_QUEUE = "polar-sync"; + +/** Hourly Oura Cloud poll (:15). */ +export const OURA_SYNC_QUEUE = "oura-sync"; + +/** Hourly Strava poll (:17). */ +export const STRAVA_SYNC_QUEUE = "strava-sync"; diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index eaaf55544..7abc44a0e 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -77,6 +77,12 @@ import { type ScheduleEntry, } from "./registrar-shared"; import { WITHINGS_ECG_SYNC_QUEUE } from "@/lib/jobs/withings-ecg-queue"; +import { + NIGHTSCOUT_SYNC_QUEUE, + OURA_SYNC_QUEUE, + POLAR_SYNC_QUEUE, + STRAVA_SYNC_QUEUE, +} from "@/lib/jobs/integration-poll-queues"; import { WithingsSyncPayload, WithingsActivitySyncPayload, @@ -223,8 +229,11 @@ const GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_CRON = "26 3 * * *"; // pulls the recent SGV window per configured instance. :11 staggers off the // WHOOP (:05), Fitbit (:08), and Withings sync ticks so the hourly polls don't // pile up on one boss poll. -const NIGHTSCOUT_SYNC_QUEUE = "nightscout-sync"; - +// +// The four poll-queue NAMES live in `@/lib/jobs/integration-poll-queues` (see +// the import above) so an OAuth callback can enqueue a connect-time first sync +// without importing this registrar and its whole handler graph. Creation, +// scheduling, and handler binding stay here. const NIGHTSCOUT_SYNC_CRON = "11 * * * *"; // every hour at :11 // v1.17.0 (F4) — Polar + Oura OAuth poll sync. Poll-only (one hourly tick per // provider re-walks every connected user). :13 / :15 stagger off the other @@ -232,16 +241,13 @@ const NIGHTSCOUT_SYNC_CRON = "11 * * * *"; // every hour at :11 // pile up on one boss poll. The queues MUST be registered in `allQueues` below // or pg-boss never provisions them and the schedule silently no-ops (the // v1.4.37 dead-queue class). -const POLAR_SYNC_QUEUE = "polar-sync"; const POLAR_SYNC_CRON = "13 * * * *"; // every hour at :13 -const OURA_SYNC_QUEUE = "oura-sync"; const OURA_SYNC_CRON = "15 * * * *"; // every hour at :15 // v1.28.x — Strava OAuth poll sync. Poll-only (webhook deferred): one hourly // tick re-walks every connected user. :17 staggers off the Polar (:13) / Oura // (:15) ticks so the polls don't pile up on one boss poll. The queue MUST be // registered in `allQueues` below or pg-boss never provisions it and the // schedule silently no-ops (the v1.4.37 dead-queue class). -const STRAVA_SYNC_QUEUE = "strava-sync"; const STRAVA_SYNC_CRON = "17 * * * *"; // every hour at :17 // v1.30.x — daily sweep for the native OIDC handoff ledger. Handoff codes are // single-use + 90s-lived and expiry is enforced at read, so this is pure