feat(fitbit): add Google OAuth, credentials, and token refresh

Land the OAuth + credentials half of the Fitbit / Google Health provider,
cloning the WHOOP provider structure and adapting it to the Google identity
flow.

- Client OAuth half (src/lib/fitbit/client.ts): Google authorize-URL builder
  (access_type=offline + prompt=consent), authorization-code exchange, token
  refresh, and a profile fetch for the connection's external user id. Client
  credentials ride in an HTTP Basic-auth header on the token endpoint; the
  four launch Restricted read scopes are pinned in FITBIT_OAUTH_SCOPE. All
  egress goes through safeFetch.
- Non-rotating refresh tokens: getValidToken refreshes at expiry-minus-five-
  minutes, persists the new access token + expiry, and overwrites the stored
  refresh token ONLY when the response carries a fresh one — Google keeps the
  refresh token stable, unlike WHOOP.
- BYO credentials route (PUT/GET/DELETE /api/fitbit/credentials): per-user
  client id/secret stored encrypted on the User columns, built field-by-field
  under requireAuth.
- OAuth state + connect + callback routes: a random base64url nonce backed by
  a 10-minute FitbitOAuthState ledger row carries the (nonce -> userId)
  mapping; the callback validates the cookie/URL state with timingSafeEqual,
  consumes the row atomically (replay/expired/cross-user reason tags),
  exchanges the code, upserts the encrypted FitbitConnection, clears prior
  reauth state, and enqueues the self-converging history backfill.
- Response classifier mirrors WHOOP's HTTP-status buckets; Google signals a
  revoked grant with a 401, so there is no invalid_grant body special-case.

Tests cover the authorize URL, Basic-auth token exchange, the no-rotation
refresh branch (token preserved when omitted, overwritten when present), the
classifier status table, and the OAuth-state nonce mint.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Marc-André BombeckandClaude Opus 4.8 committed 2026-06-04 21:49:39 +02:00
1 parent d4801d4646
commit 7ab2cbaa56
14 files changed
+1561

No files matched your search

+176
View File
@@ -0,0 +1,176 @@
import { prisma } from "@/lib/db";
import { apiHandler, requireAuth } from "@/lib/api-handler";
import { annotate, getEvent } from "@/lib/logging/context";
import { auditLog } from "@/lib/auth/audit";
import { encrypt } from "@/lib/crypto";
import {
exchangeCode,
fetchProfile,
resolveFitbitUserId,
FITBIT_OAUTH_SCOPE,
} from "@/lib/fitbit/client";
import { getUserFitbitCredentials } from "@/lib/fitbit/credentials";
import { FITBIT_OAUTH_STATE_COOKIE } from "@/lib/fitbit/oauth-state";
import { FITBIT_BACKFILL_QUEUE } from "@/lib/jobs/fitbit-backfill";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { markReconnected } from "@/lib/integrations/status";
import { Prisma } from "@/generated/prisma/client";
import { NextRequest, NextResponse } from "next/server";
import { timingSafeEqual } from "node:crypto";
/**
* OAuth callback from Google for Fitbit / Google Health (v1.12.0). Mirrors the
* WHOOP callback: the `state` param is a random base64url nonce keyed against
* the `FitbitOAuthState` ledger. The in-flight user is resolved via the row's
* `userId` (never parsed from the cookie value) and cross-checked against the
* session. The row is consumed (deleted) atomically on every exit branch so a
* replay of the same nonce fails the second time.
*
* Reason tags distinguish the four post-delete branches for the audit trail:
* `csrf1` (URL/cookie mismatch, short-circuit before delete), `replay` (P2025 —
* nonce already consumed), `expired` (valid row, TTL elapsed), and `cross_user`
* (valid row, session userId mismatch).
*
* On success: exchange the code, fetch the Google Health profile for the
* external `fitbitUserId`, persist the encrypted `FitbitConnection`, clear any
* prior reauth state, and enqueue the self-converging history backfill.
*/
const ERR = (reason: string) =>
NextResponse.redirect(
new URL(
`/settings/integrations?fitbit=error&reason=${reason}`,
process.env.NEXT_PUBLIC_APP_URL!,
),
);
export const GET = apiHandler(async (request: NextRequest) => {
const { user } = await requireAuth();
annotate({ action: { name: "fitbit.callback" } });
const { searchParams } = new URL(request.url);
const code = searchParams.get("code");
const state = searchParams.get("state");
const storedState = request.cookies.get(FITBIT_OAUTH_STATE_COOKIE)?.value;
// CSRF leg 1: URL state must match the cookie state byte-for-byte
// (timing-safe). Runs BEFORE the atomic delete so a probe with arbitrary
// state values can't grief legitimate ledger rows.
if (
!state ||
!storedState ||
state.length !== storedState.length ||
!timingSafeEqual(Buffer.from(state), Buffer.from(storedState))
) {
annotate({ meta: { reason: "csrf1" } });
return ERR("csrf1");
}
// CSRF leg 2: atomically consume the ledger row. `delete` returns the row on
// success so the `expiresAt` + `userId` checks run against the consumed
// payload. P2025 means the nonce was already consumed (replay) or never
// existed. Atomic at the Postgres row level: two concurrent callbacks with the
// same nonce can't both pass before either deletes.
let stateRow: { userId: string; expiresAt: Date } | null = null;
try {
stateRow = await prisma.fitbitOAuthState.delete({
where: { nonce: state },
});
} catch (err) {
if (
err instanceof Prisma.PrismaClientKnownRequestError &&
err.code === "P2025"
) {
annotate({ meta: { reason: "replay" } });
return ERR("replay");
}
const errName = err instanceof Error ? err.name : "unknown";
getEvent()?.addWarning(`oauth-state-delete failed: ${errName}`);
annotate({ meta: { reason: "state" } });
return ERR("state");
}
if (stateRow.expiresAt <= new Date()) {
annotate({ meta: { reason: "expired" } });
return ERR("expired");
}
if (stateRow.userId !== user.id) {
annotate({ meta: { reason: "cross_user" } });
return ERR("cross_user");
}
if (!code) {
return ERR("nocode");
}
try {
const creds = await getUserFitbitCredentials(user.id);
if (!creds) {
return ERR("nocreds");
}
const tokens = await exchangeCode(code, creds);
const profile = await fetchProfile(tokens.access_token);
const fitbitUserId = resolveFitbitUserId(profile);
const expiresAt = new Date(Date.now() + tokens.expires_in * 1000);
// The initial code exchange always returns a refresh token (the connect
// route forces `prompt=consent` + `access_type=offline`). Fall back to the
// existing stored token defensively if Google ever omits it on a re-consent.
await prisma.fitbitConnection.upsert({
where: { userId: user.id },
update: {
fitbitUserId,
accessToken: encrypt(tokens.access_token),
tokenExpiresAt: expiresAt,
scope: tokens.scope ?? FITBIT_OAUTH_SCOPE,
backfillCompletedAt: null,
...(tokens.refresh_token
? { refreshToken: encrypt(tokens.refresh_token) }
: {}),
},
create: {
userId: user.id,
fitbitUserId,
accessToken: encrypt(tokens.access_token),
refreshToken: encrypt(tokens.refresh_token ?? ""),
tokenExpiresAt: expiresAt,
scope: tokens.scope ?? FITBIT_OAUTH_SCOPE,
},
});
await auditLog("fitbit.connect", {
userId: user.id,
details: { fitbitUserId },
});
// Re-completing OAuth clears any prior reauth-required state.
await markReconnected(user.id, "fitbit");
// Enqueue the self-converging history backfill. Best-effort: the boot-time
// discovery query (`backfillCompletedAt IS NULL`) is the safety net, so a
// missing boss instance here doesn't strand the connection.
const boss = getGlobalBoss();
if (boss) {
await boss
.send(FITBIT_BACKFILL_QUEUE, {
userId: user.id,
enqueuedAt: new Date().toISOString(),
})
.catch((err) =>
getEvent()?.addWarning(`fitbit-backfill enqueue failed: ${err}`),
);
}
const response = NextResponse.redirect(
new URL(
"/settings/integrations?fitbit=connected",
process.env.NEXT_PUBLIC_APP_URL!,
),
);
response.cookies.delete(FITBIT_OAUTH_STATE_COOKIE);
return response;
} catch (err) {
getEvent()?.setError(err);
return ERR("token");
}
});
+90
View File
@@ -0,0 +1,90 @@
import { apiHandler, requireAuth } from "@/lib/api-handler";
import { apiError } from "@/lib/api-response";
import { prisma } from "@/lib/db";
import { annotate, getEvent } from "@/lib/logging/context";
import { checkRateLimit } from "@/lib/rate-limit";
import { getAuthorizationUrl } from "@/lib/fitbit/client";
import { getUserFitbitCredentials } from "@/lib/fitbit/credentials";
import {
FITBIT_OAUTH_STATE_COOKIE,
FITBIT_OAUTH_STATE_TTL_MS,
mintFitbitOAuthStateNonce,
} from "@/lib/fitbit/oauth-state";
import { NextRequest, NextResponse } from "next/server";
import { shouldEmitSecureCookie } from "@/lib/auth/secure-cookie";
/**
* Redirect the user to the Google OAuth consent page for Fitbit / Google Health
* (v1.12.0).
*
* Mirrors the WHOOP connect route: a fully-random base64url state nonce backed
* by a 10-minute `FitbitOAuthState` ledger row carries the `(nonce → userId)`
* mapping, so the user id never travels in the OAuth `state` param (which can
* land in request logs / network captures). The httpOnly + Secure cookie
* carries JUST the nonce; the callback resolves the user via the row's `userId`.
*
* Rate-limited per user (10 calls / 60 s) so a logged-in session can't spam
* ledger rows for the full 10-min TTL window. Both the rate-limit and
* create-failure paths redirect (not JSON) because the entry point is a browser
* navigation — a 429 envelope would surface as a blank page.
*/
const CONNECT_RATE_LIMIT = 10;
const CONNECT_WINDOW_MS = 60_000;
export const GET = apiHandler(async (req: NextRequest) => {
const { user } = await requireAuth();
annotate({ action: { name: "fitbit.connect" } });
const rl = await checkRateLimit(
`fitbit:connect:${user.id}`,
CONNECT_RATE_LIMIT,
CONNECT_WINDOW_MS,
);
if (!rl.allowed) {
annotate({ action: { name: "fitbit.connect.rate_limited" } });
return NextResponse.redirect(
new URL(
"/settings/integrations?fitbit=error&reason=rate_limited",
req.url,
),
);
}
const creds = await getUserFitbitCredentials(user.id);
if (!creds) {
return apiError(
"Please configure your Google OAuth Client ID and Client Secret in Settings first.",
400,
);
}
const nonce = mintFitbitOAuthStateNonce();
try {
await prisma.fitbitOAuthState.create({
data: {
nonce,
userId: user.id,
expiresAt: new Date(Date.now() + FITBIT_OAUTH_STATE_TTL_MS),
},
});
} catch (err) {
getEvent()?.setError(err);
annotate({ action: { name: "fitbit.connect.create_failed" } });
return NextResponse.redirect(
new URL("/settings/integrations?fitbit=error&reason=connect", req.url),
);
}
const url = getAuthorizationUrl(nonce, creds);
const response = NextResponse.redirect(url);
response.cookies.set(FITBIT_OAUTH_STATE_COOKIE, nonce, {
httpOnly: true,
secure: shouldEmitSecureCookie(),
sameSite: "lax",
maxAge: Math.floor(FITBIT_OAUTH_STATE_TTL_MS / 1000),
path: "/",
});
return response;
});
+79
View File
@@ -0,0 +1,79 @@
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 { encrypt } from "@/lib/crypto";
import { fitbitCredentialsSchema } from "@/lib/validations/fitbit";
import { NextRequest } from "next/server";
import { z } from "zod/v4";
/**
* Check whether the user has Fitbit / Google Health BYO-key credentials stored
* (v1.12.0).
*/
export const GET = apiHandler(async () => {
const { user } = await requireAuth();
annotate({ action: { name: "fitbit.credentials.get" } });
const dbUser = await prisma.user.findUnique({
where: { id: user.id },
select: {
fitbitClientIdEncrypted: true,
fitbitClientSecretEncrypted: true,
},
});
return apiSuccess({
hasCredentials:
!!dbUser?.fitbitClientIdEncrypted &&
!!dbUser?.fitbitClientSecretEncrypted,
});
});
/**
* Save Fitbit / Google Health API credentials (encrypted at rest).
*/
export const PUT = apiHandler(async (request: NextRequest) => {
const { user } = await requireAuth();
annotate({ action: { name: "fitbit.credentials.update" } });
const { data: body, error: jsonError } = await safeJson(request);
if (jsonError) return jsonError;
const result = z.safeParse(fitbitCredentialsSchema, body);
if (!result.success) {
return apiError("Client ID and Client Secret are required", 422);
}
await prisma.user.update({
where: { id: user.id },
data: {
fitbitClientIdEncrypted: encrypt(result.data.clientId),
fitbitClientSecretEncrypted: encrypt(result.data.clientSecret),
},
});
return apiSuccess({ updated: true });
});
/**
* Delete Fitbit credentials and the active connection.
*/
export const DELETE = apiHandler(async () => {
const { user } = await requireAuth();
annotate({ action: { name: "fitbit.credentials.delete" } });
await prisma.fitbitConnection
.delete({ where: { userId: user.id } })
.catch(() => {});
await prisma.user.update({
where: { id: user.id },
data: {
fitbitClientIdEncrypted: null,
fitbitClientSecretEncrypted: null,
},
});
return apiSuccess({ deleted: true });
});
+162
View File
@@ -0,0 +1,162 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import {
FITBIT_API_BASE,
FITBIT_OAUTH_SCOPE,
exchangeCode,
fetchProfile,
getAuthorizationUrl,
refreshAccessToken,
resolveFitbitUserId,
} from "../client";
const CREDS = { clientId: "cid", clientSecret: "csecret" };
/** Stub global fetch with a queue of `{ status, body }` responses. */
function installFetchMock(pages: Array<{ status: number; body: unknown }>) {
let i = 0;
const fetchMock = vi.fn(async () => {
const page = pages[Math.min(i, pages.length - 1)]!;
i += 1;
return {
status: page.status,
json: async () => page.body,
};
});
vi.stubGlobal("fetch", fetchMock);
return fetchMock;
}
afterEach(() => {
vi.unstubAllGlobals();
vi.restoreAllMocks();
});
describe("getAuthorizationUrl", () => {
it("builds the Google authorize URL with offline access + consent prompt", () => {
const url = getAuthorizationUrl("nonce123", CREDS);
expect(url).toContain("accounts.google.com/o/oauth2/v2/auth");
expect(url).toContain("response_type=code");
expect(url).toContain("client_id=cid");
expect(url).toContain("state=nonce123");
expect(url).toContain("access_type=offline");
expect(url).toContain("prompt=consent");
// URLSearchParams encodes the space-separated scope; compare the parsed
// `scope` param back to the canonical constant.
const scope = new URL(url).searchParams.get("scope");
expect(scope).toBe(FITBIT_OAUTH_SCOPE);
expect(FITBIT_OAUTH_SCOPE).toContain("googlehealth.profile.readonly");
});
it("requests exactly the four launch Restricted read bundles", () => {
const scopes = FITBIT_OAUTH_SCOPE.split(" ");
expect(scopes).toHaveLength(4);
expect(scopes).toContain(
"https://www.googleapis.com/auth/googlehealth.activity_and_fitness.readonly",
);
expect(scopes).toContain(
"https://www.googleapis.com/auth/googlehealth.health_metrics_and_measurements.readonly",
);
expect(scopes).toContain(
"https://www.googleapis.com/auth/googlehealth.sleep.readonly",
);
// ECG / nutrition / location are deliberately omitted from the launch set.
expect(FITBIT_OAUTH_SCOPE).not.toContain("ecg");
expect(FITBIT_OAUTH_SCOPE).not.toContain("nutrition");
expect(FITBIT_OAUTH_SCOPE).not.toContain("location");
});
});
describe("token exchange + refresh", () => {
it("exchanges an authorization code for a token pair via Basic auth", async () => {
const fetchMock = installFetchMock([
{
status: 200,
body: {
access_token: "at",
refresh_token: "rt",
expires_in: 3600,
scope: "https://www.googleapis.com/auth/googlehealth.sleep.readonly",
},
},
]);
const tok = await exchangeCode("code", CREDS);
expect(tok.access_token).toBe("at");
expect(tok.refresh_token).toBe("rt");
expect(tok.expires_in).toBe(3600);
const [url, init] = fetchMock.mock.calls[0] as unknown as [
string,
{ headers: Record<string, string>; body: string },
];
expect(url).toContain("oauth2.googleapis.com/token");
expect(init.body).toContain("grant_type=authorization_code");
// Confidential client credentials ride in the Basic-auth header.
const expected = `Basic ${Buffer.from("cid:csecret").toString("base64")}`;
expect(init.headers.Authorization).toBe(expected);
});
it("refreshes WITHOUT re-sending scope (Google preserves the grant)", async () => {
const fetchMock = installFetchMock([
{
status: 200,
body: { access_token: "at2", expires_in: 3600 },
},
]);
const tok = await refreshAccessToken("rt1", CREDS);
expect(tok.access_token).toBe("at2");
const [, init] = fetchMock.mock.calls[0] as unknown as [
string,
{ body: string },
];
expect(init.body).toContain("grant_type=refresh_token");
expect(init.body).toContain("refresh_token=rt1");
// Google preserves the original grant — no scope param is re-sent.
expect(init.body).not.toContain("scope=");
});
it("returns an absent refresh_token unchanged (Google does not rotate)", async () => {
installFetchMock([
{ status: 200, body: { access_token: "at3", expires_in: 3600 } },
]);
const tok = await refreshAccessToken("rt1", CREDS);
expect(tok.access_token).toBe("at3");
expect(tok.refresh_token).toBeUndefined();
});
it("throws a classified FitbitApiError on a 401 token response", async () => {
installFetchMock([{ status: 401, body: { error: "invalid_grant" } }]);
await expect(exchangeCode("bad", CREDS)).rejects.toMatchObject({
name: "FitbitApiError",
classification: "reauth_required",
});
});
});
describe("fetchProfile + resolveFitbitUserId", () => {
it("fetches the profile from the Google Health base", async () => {
const fetchMock = installFetchMock([
{ status: 200, body: { name: "users/abc123" } },
]);
const profile = await fetchProfile("at");
expect(profile.name).toBe("users/abc123");
const [url] = fetchMock.mock.calls[0] as unknown as [string];
expect(url).toBe(`${FITBIT_API_BASE}/users/me/profile`);
});
it("throws a classified error on a 403 profile read", async () => {
installFetchMock([{ status: 403, body: {} }]);
await expect(fetchProfile("at")).rejects.toMatchObject({
name: "FitbitApiError",
classification: "reauth_required",
});
});
it("resolves the external user id from a users/{id} resource name", () => {
expect(resolveFitbitUserId({ name: "users/abc123" })).toBe("abc123");
});
it("falls back to a bare id, then to 'me'", () => {
expect(resolveFitbitUserId({ id: "xyz" })).toBe("xyz");
expect(resolveFitbitUserId({})).toBe("me");
});
});
@@ -0,0 +1,26 @@
import { describe, expect, it } from "vitest";
import {
FITBIT_OAUTH_STATE_COOKIE,
FITBIT_OAUTH_STATE_TTL_MS,
mintFitbitOAuthStateNonce,
} from "../oauth-state";
describe("Fitbit OAuth state", () => {
it("pins the cookie name and 10-minute TTL", () => {
expect(FITBIT_OAUTH_STATE_COOKIE).toBe("fitbit_state");
expect(FITBIT_OAUTH_STATE_TTL_MS).toBe(10 * 60 * 1000);
});
it("mints a 22-char base64url nonce (128 bits, no padding)", () => {
const nonce = mintFitbitOAuthStateNonce();
expect(nonce).toHaveLength(22);
expect(nonce).toMatch(/^[A-Za-z0-9_-]{22}$/);
});
it("mints unique nonces", () => {
const set = new Set(
Array.from({ length: 100 }, () => mintFitbitOAuthStateNonce()),
);
expect(set.size).toBe(100);
});
});
@@ -0,0 +1,100 @@
import { describe, expect, it } from "vitest";
import {
FitbitApiError,
classifyFitbitError,
classifyFitbitResponse,
} from "../response-classifier";
describe("classifyFitbitResponse", () => {
it("classifies 2xx as success", () => {
for (const status of [200, 201, 204, 299]) {
const v = classifyFitbitResponse(status);
expect(v.classification).toBe("success");
expect(v.httpStatus).toBe(status);
}
});
it("classifies 429 as transient (rate-limit)", () => {
const v = classifyFitbitResponse(429);
expect(v.classification).toBe("transient");
expect(v.reason).toBe("http_429");
});
it("classifies 401 and 403 as reauth_required", () => {
expect(classifyFitbitResponse(401).classification).toBe("reauth_required");
expect(classifyFitbitResponse(403).classification).toBe("reauth_required");
});
it("classifies other 4xx as persistent", () => {
for (const status of [400, 404, 422]) {
expect(classifyFitbitResponse(status).classification).toBe("persistent");
}
});
it("classifies 5xx as transient", () => {
for (const status of [500, 502, 503, 504]) {
expect(classifyFitbitResponse(status).classification).toBe("transient");
}
});
it("defaults an unrecognised status to transient", () => {
expect(classifyFitbitResponse(308).classification).toBe("transient");
});
});
describe("FitbitApiError", () => {
it("carries the verdict and caps the message at 1024 chars", () => {
const err = new FitbitApiError({
verb: "exchangeCode",
classification: "reauth_required",
httpStatus: 401,
reason: "http_401",
upstreamError: "invalid_grant",
});
expect(err.name).toBe("FitbitApiError");
expect(err.classification).toBe("reauth_required");
expect(err.httpStatus).toBe(401);
expect(err.upstreamError).toBe("invalid_grant");
expect(err.message).toContain("invalid_grant");
expect(err.message.length).toBeLessThanOrEqual(1024);
});
});
describe("classifyFitbitError", () => {
it("reads the classification off a FitbitApiError", () => {
const err = new FitbitApiError({
verb: "refreshAccessToken",
classification: "persistent",
httpStatus: 400,
reason: "http_400",
});
expect(classifyFitbitError(err)).toBe("persistent");
});
it("parses the status out of a legacy message shape", () => {
expect(classifyFitbitError(new Error("Fitbit fetchProfile error: 401"))).toBe(
"reauth_required",
);
expect(classifyFitbitError(new Error("Fitbit sync error: 500"))).toBe(
"transient",
);
});
it("defaults a non-Fitbit error to transient", () => {
expect(classifyFitbitError(new Error("network down"))).toBe("transient");
expect(classifyFitbitError("boom")).toBe("transient");
});
it("treats a 401 on the token endpoint as reauth (no invalid_grant special-case)", () => {
// Google signals a revoked grant with a 401 status, not a 400 invalid_grant
// body — the status-driven classifier already buckets it as reauth.
const err = new FitbitApiError({
verb: "refreshAccessToken",
classification: "reauth_required",
httpStatus: 401,
reason: "http_401",
upstreamError: "invalid_grant",
});
expect(classifyFitbitError(err)).toBe("reauth_required");
});
});
+194
View File
@@ -0,0 +1,194 @@
/**
* v1.12.0 — Fitbit token-management tests (mocked). Covers `getValidToken`:
* - the NO-ROTATION refresh branch (refresh token preserved when the response
* omits it; overwritten only when a fresh one is returned);
* - the stored-token fast path when not near expiry;
* - a reauth failure recorded when credentials are missing on refresh.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
const { prismaMock, refreshAccessTokenMock, recordSyncFailure } = vi.hoisted(
() => ({
prismaMock: {
fitbitConnection: {
findUnique: vi.fn(),
update: vi.fn(),
},
},
refreshAccessTokenMock: vi.fn(),
recordSyncFailure: vi.fn<(...a: unknown[]) => Promise<void>>(async () => {}),
}),
);
vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
vi.mock("@/lib/crypto", () => ({
encrypt: (s: string) => `enc(${s})`,
decrypt: (s: string) => s.replace(/^enc\(|\)$/g, ""),
}));
vi.mock("../client", async (orig) => {
const actual = await orig<typeof import("../client")>();
return {
...actual,
refreshAccessToken: (...a: unknown[]) => refreshAccessTokenMock(...a),
};
});
vi.mock("../credentials", () => ({
getUserFitbitCredentials: vi.fn(async () => ({
clientId: "cid",
clientSecret: "csecret",
})),
}));
vi.mock("@/lib/integrations/status", () => ({
recordSyncFailure: (...a: unknown[]) => recordSyncFailure(...a),
}));
import {
classificationToFailureKind,
getValidToken,
} from "../sync";
beforeEach(() => {
vi.clearAllMocks();
});
afterEach(() => {
vi.clearAllMocks();
});
describe("getValidToken — non-rotating refresh", () => {
it("persists the new access token and KEEPS the stored refresh token when the response omits one", async () => {
prismaMock.fitbitConnection.findUnique.mockResolvedValue({
id: "conn1",
fitbitUserId: "abc123",
accessToken: "enc(old-access)",
refreshToken: "enc(old-refresh)",
// Expired so the refresh path fires.
tokenExpiresAt: new Date(Date.now() - 1000),
});
// Google does NOT return a refresh_token on a routine refresh.
refreshAccessTokenMock.mockResolvedValue({
access_token: "new-access",
expires_in: 3600,
});
prismaMock.fitbitConnection.update.mockResolvedValue({});
const result = await getValidToken("user1");
expect(result?.accessToken).toBe("new-access");
expect(refreshAccessTokenMock).toHaveBeenCalledWith("old-refresh", {
clientId: "cid",
clientSecret: "csecret",
});
const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0];
expect(updateArg.data.accessToken).toBe("enc(new-access)");
// The stored refresh token must NOT be touched when the response omits one.
expect(updateArg.data).not.toHaveProperty("refreshToken");
});
it("overwrites the stored refresh token only when the response carries a fresh one", async () => {
prismaMock.fitbitConnection.findUnique.mockResolvedValue({
id: "conn1",
fitbitUserId: "abc123",
accessToken: "enc(old-access)",
refreshToken: "enc(old-refresh)",
tokenExpiresAt: new Date(Date.now() - 1000),
});
refreshAccessTokenMock.mockResolvedValue({
access_token: "new-access",
refresh_token: "rotated-refresh",
expires_in: 3600,
});
prismaMock.fitbitConnection.update.mockResolvedValue({});
await getValidToken("user1");
const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0];
expect(updateArg.data.accessToken).toBe("enc(new-access)");
expect(updateArg.data.refreshToken).toBe("enc(rotated-refresh)");
});
it("returns the stored token without refresh when not near expiry", async () => {
prismaMock.fitbitConnection.findUnique.mockResolvedValue({
id: "conn1",
fitbitUserId: "abc123",
accessToken: "enc(live-access)",
refreshToken: "enc(live-refresh)",
tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
});
const result = await getValidToken("user1");
expect(result?.accessToken).toBe("live-access");
expect(refreshAccessTokenMock).not.toHaveBeenCalled();
expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled();
});
it("returns null when there is no connection", async () => {
prismaMock.fitbitConnection.findUnique.mockResolvedValue(null);
expect(await getValidToken("user1")).toBeNull();
});
it("records a reauth failure when credentials are missing on refresh", async () => {
const { getUserFitbitCredentials } = await import("../credentials");
(
getUserFitbitCredentials as ReturnType<typeof vi.fn>
).mockResolvedValueOnce(null);
prismaMock.fitbitConnection.findUnique.mockResolvedValue({
id: "conn1",
fitbitUserId: "abc123",
accessToken: "enc(a)",
refreshToken: "enc(r)",
tokenExpiresAt: new Date(Date.now() - 1000),
});
const result = await getValidToken("user1");
expect(result).toBeNull();
expect(recordSyncFailure).toHaveBeenCalledWith(
expect.objectContaining({
integration: "fitbit",
kind: "reauth_required",
}),
);
});
it("records a classified failure and returns null when the refresh itself fails", async () => {
prismaMock.fitbitConnection.findUnique.mockResolvedValue({
id: "conn1",
fitbitUserId: "abc123",
accessToken: "enc(a)",
refreshToken: "enc(r)",
tokenExpiresAt: new Date(Date.now() - 1000),
});
refreshAccessTokenMock.mockRejectedValue(
new Error("Fitbit refreshAccessToken error: 401"),
);
const result = await getValidToken("user1");
expect(result).toBeNull();
expect(recordSyncFailure).toHaveBeenCalledWith(
expect.objectContaining({
integration: "fitbit",
kind: "reauth_required",
}),
);
});
});
describe("classificationToFailureKind", () => {
it("maps each classification onto a FailureKind", () => {
expect(classificationToFailureKind("reauth_required")).toBe(
"reauth_required",
);
expect(classificationToFailureKind("persistent")).toBe("persistent");
expect(classificationToFailureKind("transient")).toBe("transient");
// A success has no failure kind; surface it as transient (contract-bug
// anomaly is still recorded).
expect(classificationToFailureKind("success")).toBe("transient");
});
});
+235
View File
@@ -0,0 +1,235 @@
/**
* Fitbit / Google Health API client — OAuth half (v1.12.0).
* Docs: https://developers.google.com/health (re-verify at build — the Google
* Health API is young; v4, post-Fitbit-Web-API).
*
* Mirrors the WHOOP client structure (`src/lib/whoop/client.ts`): hand-rolled
* fetch over `safeFetch` (no SDK), an OAuth handshake (`getAuthorizationUrl` /
* `exchangeCode` / `refreshAccessToken`), and a single profile fetch for the
* connection's external user id. The data-fetch half (paginated `dataPoints`
* walker + per-type mappers + `FITBIT_FIELD_MAP`) lands in a later wave.
*
* KEY DELTA vs WHOOP — Google does NOT rotate refresh tokens. WHOOP invalidates
* the prior refresh token on every refresh; Google's refresh tokens are stable
* (long-lived in production, 7-day expiry only in Testing mode). So
* `refreshAccessToken` may return a token response WITHOUT a `refresh_token`,
* and the sync layer keeps the stored one in that case (see `getValidToken` in
* `sync.ts`).
*
* Confidential web-server client: HealthLog holds the client secret
* server-side, so PKCE is omitted (matches WHOOP). Client credentials are sent
* to the token endpoint via HTTP Basic auth (RFC 6749 §2.3.1), the way Google's
* token endpoint accepts a confidential client.
*/
import { getEvent } from "@/lib/logging/context";
import { safeFetch } from "@/lib/safe-fetch";
import { FitbitApiError, classifyFitbitResponse } from "./response-classifier";
export const FITBIT_API_BASE = "https://health.googleapis.com/v4";
const GOOGLE_OAUTH_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth";
const GOOGLE_OAUTH_TOKEN_URL = "https://oauth2.googleapis.com/token";
export interface FitbitCredentials {
clientId: string;
clientSecret: string;
}
function getRedirectUri(): string {
return (
process.env.FITBIT_REDIRECT_URI ??
`${process.env.NEXT_PUBLIC_APP_URL}/api/fitbit/callback`
);
}
/**
* OAuth scopes HealthLog requests (space-separated on the wire). The four
* launch Restricted read bundles cover every v1.12.0 metric; ECG / nutrition /
* location / IRN are deliberately omitted to keep the Restricted-scope CASA
* review surface minimal. Every scope is Restricted → the operator's OAuth
* client needs Google brand verification + an annual CASA assessment before it
* leaves Testing mode.
*/
export const FITBIT_OAUTH_SCOPE = [
"https://www.googleapis.com/auth/googlehealth.activity_and_fitness.readonly",
"https://www.googleapis.com/auth/googlehealth.health_metrics_and_measurements.readonly",
"https://www.googleapis.com/auth/googlehealth.sleep.readonly",
"https://www.googleapis.com/auth/googlehealth.profile.readonly",
].join(" ");
/**
* Generate the Google OAuth authorization URL (a browser redirect, not a
* fetch). `state` is the opaque CSRF nonce minted by `oauth-state.ts`.
*
* `access_type=offline` + `prompt=consent` are Google's requirement to receive
* a refresh token (the equivalent of WHOOP's `offline` scope); `prompt=consent`
* forces the consent screen so a re-connect always returns a fresh refresh
* token even if the user previously granted.
*/
export function getAuthorizationUrl(
state: string,
creds: FitbitCredentials,
): string {
const params = new URLSearchParams({
response_type: "code",
client_id: creds.clientId,
redirect_uri: getRedirectUri(),
scope: FITBIT_OAUTH_SCOPE,
access_type: "offline",
prompt: "consent",
state,
});
return `${GOOGLE_OAUTH_AUTH_URL}?${params}`;
}
export interface FitbitTokenResponse {
access_token: string;
/**
* Present on the initial code exchange and whenever Google issues a new
* refresh token; ABSENT on a routine refresh because Google does not rotate.
* The sync layer keeps the stored token when this is missing.
*/
refresh_token?: string;
expires_in: number;
scope?: string;
token_type?: string;
}
/** Basic-auth header carrying the confidential client credentials. */
function basicAuthHeader(creds: FitbitCredentials): string {
const raw = `${creds.clientId}:${creds.clientSecret}`;
return `Basic ${Buffer.from(raw, "utf8").toString("base64")}`;
}
async function postToken(
params: URLSearchParams,
creds: FitbitCredentials,
verb: string,
): Promise<FitbitTokenResponse> {
const start = performance.now();
const res = await safeFetch(GOOGLE_OAUTH_TOKEN_URL, {
method: "POST",
headers: {
"Content-Type": "application/x-www-form-urlencoded",
Authorization: basicAuthHeader(creds),
},
body: params.toString(),
});
const json = await res.json().catch(() => null);
const verdict = classifyFitbitResponse(res.status);
getEvent()?.addExternalCall({
service: "fitbit",
method: verb,
duration_ms: Math.round(performance.now() - start),
status: res.status,
error: verdict.classification === "success" ? undefined : verdict.reason,
});
if (verdict.classification !== "success") {
throw new FitbitApiError({
verb,
classification: verdict.classification,
httpStatus: verdict.httpStatus,
reason: verdict.reason,
upstreamError: typeof json?.error === "string" ? json.error : undefined,
});
}
return json as FitbitTokenResponse;
}
/** Exchange an authorization code for the initial token pair. */
export async function exchangeCode(
code: string,
creds: FitbitCredentials,
): Promise<FitbitTokenResponse> {
return postToken(
new URLSearchParams({
grant_type: "authorization_code",
code,
client_id: creds.clientId,
redirect_uri: getRedirectUri(),
}),
creds,
"exchangeCode",
);
}
/**
* Refresh an expired access token. Google does NOT rotate refresh tokens — the
* response carries a fresh `access_token` + `expires_in` but usually omits
* `refresh_token`. The caller persists the new access token + expiry and keeps
* the stored refresh token unless a new one is returned. The original scope is
* preserved by Google, so no `scope` param is re-sent.
*/
export async function refreshAccessToken(
refreshToken: string,
creds: FitbitCredentials,
): Promise<FitbitTokenResponse> {
return postToken(
new URLSearchParams({
grant_type: "refresh_token",
refresh_token: refreshToken,
client_id: creds.clientId,
}),
creds,
"refreshAccessToken",
);
}
/**
* The Google Health profile carries the external user id used as the
* connection's `fitbitUserId`. Mirrors WHOOP's `fetchProfile` (a single GET,
* not a paginated collection).
*/
export interface FitbitProfile {
/**
* External user identifier. Google returns `me`-relative profile data; the
* stable id is surfaced under `name` (a `users/{id}` resource name) or `id`
* depending on the API surface — both are captured here and resolved at the
* call site. Re-verify the exact field against a live account at the data-sync
* wave.
*/
name?: string;
id?: string;
}
export async function fetchProfile(
accessToken: string,
): Promise<FitbitProfile> {
const start = performance.now();
const res = await safeFetch(`${FITBIT_API_BASE}/users/me/profile`, {
method: "GET",
headers: { Authorization: `Bearer ${accessToken}` },
});
const json = (await res.json().catch(() => null)) as FitbitProfile | null;
const verdict = classifyFitbitResponse(res.status);
getEvent()?.addExternalCall({
service: "fitbit",
method: "fetchProfile",
duration_ms: Math.round(performance.now() - start),
status: res.status,
error: verdict.classification === "success" ? undefined : verdict.reason,
});
if (verdict.classification !== "success") {
throw new FitbitApiError({
verb: "fetchProfile",
classification: verdict.classification,
httpStatus: verdict.httpStatus,
reason: verdict.reason,
});
}
return (json ?? {}) as FitbitProfile;
}
/**
* Resolve the stable external user id from a Google Health profile. The
* `users/{id}` resource name is the canonical anchor; fall back to a bare `id`
* or "me" so the connection always persists a non-empty `fitbitUserId`.
*/
export function resolveFitbitUserId(profile: FitbitProfile): string {
if (profile.name) {
const tail = profile.name.split("/").pop();
if (tail) return tail;
}
if (profile.id) return profile.id;
return "me";
}
+38
View File
@@ -0,0 +1,38 @@
/**
* Helper to retrieve per-user Fitbit / Google Health OAuth credentials from the
* database. Mirrors `src/lib/whoop/credentials.ts`.
*
* Fitbit ships per-user BYO-keys: each self-hoster registers their own Google
* Cloud OAuth client and pastes the client id/secret into Settings, stored
* encrypted on `User` (the Restricted-scope brand verification + annual CASA
* assessment is per-OAuth-client, so a single shared app is unworkable for a
* multi-operator product).
*/
import { prisma } from "@/lib/db";
import { decrypt } from "@/lib/crypto";
import type { FitbitCredentials } from "./client";
/**
* Fetch and decrypt the user's Fitbit API credentials.
* Returns null if the user has not configured credentials.
*/
export async function getUserFitbitCredentials(
userId: string,
): Promise<FitbitCredentials | null> {
const user = await prisma.user.findUnique({
where: { id: userId },
select: {
fitbitClientIdEncrypted: true,
fitbitClientSecretEncrypted: true,
},
});
if (!user?.fitbitClientIdEncrypted || !user?.fitbitClientSecretEncrypted) {
return null;
}
return {
clientId: decrypt(user.fitbitClientIdEncrypted),
clientSecret: decrypt(user.fitbitClientSecretEncrypted),
};
}
+34
View File
@@ -0,0 +1,34 @@
/**
* v1.12.0 — shared constants + mint helper for the Fitbit / Google Health OAuth
* state nonce ledger. Mirrors `src/lib/whoop/oauth-state.ts`.
*
* The nonce is random + opaque; a short-lived `FitbitOAuthState` row carries
* the `(nonce → userId)` mapping so the user id never leaks into request logs
* or network captures via the OAuth `state` param. The connect route, the
* callback route, and the cleanup cron all read these constants so the
* contract has one source of truth and the unit tests can pin it without
* importing either route file.
*/
import { randomBytes } from "node:crypto";
/**
* Cookie name carrying the in-flight state nonce. httpOnly + Secure +
* sameSite:lax at the connect route (mirrors the WHOOP cookie shape).
*/
export const FITBIT_OAUTH_STATE_COOKIE = "fitbit_state" as const;
/**
* 10-minute TTL on the state row. Matches the cookie `maxAge`. Long enough to
* cover a user approving the Google consent prompt; short enough that an
* abandoned handshake doesn't strand a row for hours.
*/
export const FITBIT_OAUTH_STATE_TTL_MS = 10 * 60 * 1000;
/**
* Mint a fresh state nonce. 16 random bytes → 22 base64url chars without
* padding — 128 bits of entropy, aligned with the OAuth 2.0 §10.10
* "at least 128 bits" recommendation.
*/
export function mintFitbitOAuthStateNonce(): string {
return randomBytes(16).toString("base64url");
}
+160
View File
@@ -0,0 +1,160 @@
/**
* Off-response classifier for Fitbit / Google Health API replies (v1.12.0).
*
* Mirrors the WHOOP classifier (`src/lib/whoop/response-classifier.ts`) but
* speaks the Google Health wire. Like WHOOP, Google signals failure through the
* HTTP status code itself (no `status` field embedded in a 200 body), so this
* classifier is HTTP-status-driven.
*
* Classification buckets:
*
* - `success` → HTTP 2xx
* - `transient` → retryable: HTTP 429 (rate-limited — honour
* `Retry-After` for backoff at the call site), 5xx
* upstream outages, 3xx CDN/network glitches, and an
* empty / off-spec body where we never reached the
* envelope.
* - `reauth_required` → HTTP 401 / 403 — the access token is rejected or the
* grant was revoked; the user must redo OAuth. (The
* per-data-class 403 soft-skip lives in `sync.ts`'s
* `isCollectionForbidden`, not here, so a Restricted
* scope granted independently doesn't park the whole
* connection.)
* - `persistent` → any other 4xx (400, 404, 422, …) — a malformed
* request or a contract mismatch that retrying without
* operator intervention will not fix.
*
* The classifier is conservative: anything it doesn't recognise defaults to
* `transient` so a single unknown response doesn't hard-disable the
* integration. The 3-strike admin-alert ladder in `recordSyncFailure` already
* catches the "keeps happening" case for recurring transients.
*
* Unlike WHOOP, there is no `invalid_grant` token-body special-case: Google
* signals a revoked refresh token with a 401 on the token endpoint, which the
* status-driven classifier already buckets as `reauth_required`.
*/
/** Outcome buckets for a Fitbit response. Same shape as WHOOP. */
export type FitbitClassification =
| "success"
| "transient"
| "reauth_required"
| "persistent";
export interface ClassifiedFitbitResponse {
classification: FitbitClassification;
/** The HTTP status code that drove the verdict (undefined if none seen). */
httpStatus: number | undefined;
/** Short human-readable label for logs / audit details. */
reason: string;
}
/**
* Classify a single Fitbit / Google Health API response by its HTTP status.
*
* @param httpStatus The HTTP status from fetch.
*/
export function classifyFitbitResponse(
httpStatus: number,
): ClassifiedFitbitResponse {
if (httpStatus >= 200 && httpStatus < 300) {
return { classification: "success", httpStatus, reason: "ok" };
}
// 429 is rate-limit — honour `Retry-After` for backoff at the call site, but
// the verdict is a plain transient so the next sync retries.
if (httpStatus === 429) {
return { classification: "transient", httpStatus, reason: "http_429" };
}
// 401/403 — the bearer token is rejected or the grant revoked. The user has
// to reconnect; do not retry silently. (A per-data-class 403 soft-skip is
// applied in sync.ts before this verdict drives a connection-wide reauth.)
if (httpStatus === 401 || httpStatus === 403) {
return {
classification: "reauth_required",
httpStatus,
reason: `http_${httpStatus}`,
};
}
// Any other 4xx is a malformed request / contract mismatch — persistent.
if (httpStatus >= 400 && httpStatus < 500) {
return {
classification: "persistent",
httpStatus,
reason: `http_${httpStatus}`,
};
}
// 5xx upstream outage — retryable.
if (httpStatus >= 500 && httpStatus < 600) {
return {
classification: "transient",
httpStatus,
reason: `http_${httpStatus}`,
};
}
// 3xx (a CDN / redirect glitch — safeFetch uses redirect:"manual", so a
// surfaced 3xx is unexpected) and everything else defaults to transient.
return {
classification: "transient",
httpStatus,
reason: `http_${httpStatus}_unknown`,
};
}
/**
* Typed Error subclass carrying the classification verdict so downstream
* `try/catch` blocks can branch without re-parsing the message. Mirrors
* `WhoopApiError`.
*/
export class FitbitApiError extends Error {
readonly classification: FitbitClassification;
readonly httpStatus: number | undefined;
readonly reason: string;
readonly verb: string;
/**
* The OAuth `error` code from a token-endpoint failure body (e.g.
* `invalid_grant`). Captured for the audit trail; the classifier does not
* branch on it (Google signals a revoked grant with a 401 status).
*/
readonly upstreamError: string | undefined;
constructor(opts: {
verb: string;
classification: FitbitClassification;
httpStatus: number | undefined;
reason: string;
upstreamError?: string;
}) {
const statusLabel =
typeof opts.httpStatus === "number" ? opts.httpStatus : "?";
const errSegment = opts.upstreamError ? ` - ${opts.upstreamError}` : "";
// Cap the message at 1024 chars in the constructor (same bound as
// WhoopApiError) so a misbehaving upstream returning a multi-MB error body
// can't bloat an AuditLog row downstream.
const raw = `Fitbit ${opts.verb} error: ${statusLabel}${errSegment}`;
super(raw.slice(0, 1024));
this.name = "FitbitApiError";
this.verb = opts.verb;
this.classification = opts.classification;
this.httpStatus = opts.httpStatus;
this.reason = opts.reason;
this.upstreamError = opts.upstreamError;
}
}
/**
* Read the classification from any caught error. Returns `transient` for a
* non-`FitbitApiError` input so a call site surfacing a plain `Error` retries
* rather than permanently disabling the integration. Falls back to parsing the
* HTTP status out of the legacy `"Fitbit <verb> error: <status>"` message shape
* for callers that lose the original prototype across a pg-boss retry.
*/
export function classifyFitbitError(err: unknown): FitbitClassification {
if (err instanceof FitbitApiError) return err.classification;
const msg = err instanceof Error ? err.message : String(err);
const m = /Fitbit\s+\w+\s+error:\s*(\d+)/.exec(msg);
if (!m) return "transient";
const status = Number.parseInt(m[1]!, 10);
if (!Number.isFinite(status)) return "transient";
return classifyFitbitResponse(status).classification;
}
+158
View File
@@ -0,0 +1,158 @@
/**
* Fitbit / Google Health sync service (v1.12.0).
*
* This wave (OAuth + credentials) lands the token-management half:
* - `getValidToken` decrypts the stored token, refreshes at
* `tokenExpiresAt - 5 min`, persists the new access token + expiry, and —
* UNLIKE WHOOP — only overwrites the stored refresh token when the response
* carries a fresh one (Google does not rotate refresh tokens).
* - `recordFitbitSyncFailure` / `classificationToFailureKind` map a
* classified API error onto the shared integration-status ledger.
*
* The per-resource data sync (`upsertFitbitMeasurements`, `syncUserFitbit`, the
* collection walkers, and the 403 soft-skip orchestration) lands in a later
* wave and extends this file.
*/
import { prisma } from "@/lib/db";
import { encrypt, decrypt } from "@/lib/crypto";
import { getEvent } from "@/lib/logging/context";
import {
recordSyncFailure,
type FailureKind,
} from "@/lib/integrations/status";
import { refreshAccessToken } from "./client";
import { getUserFitbitCredentials } from "./credentials";
import {
FitbitApiError,
classifyFitbitError,
type FitbitClassification,
} from "./response-classifier";
/** Refresh the access token this many ms before `tokenExpiresAt`. */
const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000;
export interface FitbitTokenInfo {
accessToken: string;
connection: { id: string; fitbitUserId: string };
}
/**
* Resolve a valid Fitbit access token for a user, refreshing if it is within
* the 5-minute expiry buffer. Returns null when there is no connection, no
* credentials, or the refresh fails (the failure is recorded so scheduled syncs
* back off).
*
* KEY DELTA vs WHOOP: Google does not rotate refresh tokens. On refresh, persist
* the new access token + expiry and overwrite the stored `refreshToken` ONLY
* when the response carries a fresh one — otherwise keep the existing refresh
* token so the next refresh still authenticates.
*/
export async function getValidToken(
userId: string,
): Promise<FitbitTokenInfo | null> {
const connection = await prisma.fitbitConnection.findUnique({
where: { userId },
});
if (!connection) return null;
const accessToken = decrypt(connection.accessToken);
const refreshToken = decrypt(connection.refreshToken);
if (
connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS <
Date.now()
) {
try {
const creds = await getUserFitbitCredentials(userId);
if (!creds) {
getEvent()?.addWarning(
`No Fitbit credentials found for user ${userId} during token refresh`,
);
await recordSyncFailure({
userId,
integration: "fitbit",
kind: "reauth_required",
message: "Fitbit credentials missing — token refresh skipped",
errorCode: "credentials_missing",
});
return null;
}
const newTokens = await refreshAccessToken(refreshToken, creds);
const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000);
// Google does NOT rotate refresh tokens: a routine refresh returns a new
// access token but usually omits `refresh_token`. Persist the new access
// token + expiry and only overwrite the stored refresh token when the
// response actually carries a fresh one — otherwise keep the existing one.
await prisma.fitbitConnection.update({
where: { id: connection.id },
data: {
accessToken: encrypt(newTokens.access_token),
tokenExpiresAt: expiresAt,
...(newTokens.refresh_token
? { refreshToken: encrypt(newTokens.refresh_token) }
: {}),
},
});
return {
accessToken: newTokens.access_token,
connection: {
id: connection.id,
fitbitUserId: connection.fitbitUserId,
},
};
} catch (err) {
getEvent()?.addWarning(
`Fitbit token refresh failed for user ${userId}: ${err}`,
);
await recordFitbitSyncFailure(userId, err);
return null;
}
}
return {
accessToken,
connection: {
id: connection.id,
fitbitUserId: connection.fitbitUserId,
},
};
}
/**
* Map a Fitbit response classification onto a `FailureKind` and record it.
* Shared by the token-refresh path and (later) every per-resource catch block.
*/
export async function recordFitbitSyncFailure(
userId: string,
err: unknown,
): Promise<void> {
const message = err instanceof Error ? err.message : String(err);
await recordSyncFailure({
userId,
integration: "fitbit",
kind: classificationToFailureKind(classifyFitbitError(err)),
message,
errorCode:
err instanceof FitbitApiError ? err.httpStatus?.toString() : undefined,
});
}
export function classificationToFailureKind(
classification: FitbitClassification,
): FailureKind {
switch (classification) {
case "reauth_required":
return "reauth_required";
case "persistent":
return "persistent";
case "transient":
return "transient";
case "success":
// A caller asking for the FailureKind of a success is a contract bug;
// surface it as transient so the audit log still records the anomaly.
return "transient";
}
}
+94
View File
@@ -0,0 +1,94 @@
/**
* v1.12.0 — pg-boss queue + boot-time self-converging backfill for newly
* connected Fitbit / Google Health accounts. Modelled on the WHOOP backfill
* (`src/lib/jobs/whoop-backfill.ts`): a discovery query enqueues one job per
* connection that has NOT yet been backfilled, and the pass is idempotent
* across reboots (the predicate `backfill_completed_at IS NULL` drops a
* connection once its backfill finishes).
*
* The per-user backfill handler (`runFitbitBackfillForUser`) and the
* reminder-worker registration land in the data-sync wave; this module exports
* the queue contract + the discovery enqueue so the OAuth callback can request
* a backfill on connect.
*
* 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 { getGlobalBoss } from "@/lib/jobs/boss-instance";
export const FITBIT_BACKFILL_QUEUE = "fitbit-backfill";
/**
* Serial concurrency — a backfill walks years of history for one account and is
* rate-bounded by Google's per-app quota; concurrency-1 keeps it from crowding
* the request pool, matching `WHOOP_BACKFILL_CONCURRENCY`.
*/
export const FITBIT_BACKFILL_CONCURRENCY = 1;
export interface FitbitBackfillPayload {
userId: string;
enqueuedAt: string;
}
/**
* Boot-time discovery. Finds every Fitbit connection not yet backfilled
* (`backfill_completed_at IS NULL`) and enqueues one backfill job per account.
*
* Idempotent across reboots: once a connection's backfill completes,
* `backfillCompletedAt` is set and the predicate drops it from the discovery
* set. pg-boss `singletonKey` coalesces duplicate sends so a fast restart while
* a job is queued doesn't double up.
*
* Best-effort: errors are returned through the result value so the worker boot
* never fails because of a backfill miss.
*/
export async function enqueueBootTimeFitbitBackfill(): Promise<{
enqueued: number;
skipped: number;
error: string | null;
}> {
const boss = getGlobalBoss();
if (!boss) {
return { enqueued: 0, skipped: 0, error: null };
}
try {
const connections = await prisma.fitbitConnection.findMany({
where: { backfillCompletedAt: null },
select: { userId: true },
});
if (connections.length === 0) {
return { enqueued: 0, skipped: 0, error: null };
}
let enqueued = 0;
let skipped = 0;
for (const { userId } of connections) {
const payload: FitbitBackfillPayload = {
userId,
enqueuedAt: new Date().toISOString(),
};
const jobId = await boss.send(FITBIT_BACKFILL_QUEUE, payload, {
retryLimit: 3,
retryDelay: 60,
retryBackoff: true,
singletonKey: `fitbit-backfill|${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),
};
}
}
+15
View File
@@ -0,0 +1,15 @@
import { z } from "zod/v4";
/**
* Per-user Fitbit / Google Health BYO-key credentials. Each self-hoster
* registers their own Google Cloud OAuth client (the Restricted-scope brand
* verification + CASA assessment is per-OAuth-client, so a single shared app is
* unworkable for a multi-operator product) and pastes the client id/secret into
* Settings. Stored encrypted on `User`.
*/
export const fitbitCredentialsSchema = z.object({
clientId: z.string().min(1).max(200),
clientSecret: z.string().min(1).max(200),
});
export type FitbitCredentialsInput = z.infer<typeof fitbitCredentialsSchema>;