mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-13 06:32:03 +00:00
fix(withings): refresh tokens through the shared OAuth lifecycle
The status route refreshed an expiring token on its own, without the lock the sync takes, so a status read and a sync could spend the same one-time refresh token and park a healthy connection at reconnect. Both now go through the kit.
This commit is contained in:
1 parent
0592dfc074
commit
6319a1be32
4 files changed
+87
-100
No files matched your search
@@ -3,9 +3,9 @@ import { apiHandler, requireAuth } from "@/lib/api-handler";
|
||||
import { annotate, getEvent } from "@/lib/logging/context";
|
||||
import { apiSuccess } from "@/lib/api-response";
|
||||
import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load";
|
||||
import { decrypt, encrypt } from "@/lib/crypto";
|
||||
import { getUserWithingsCredentials } from "@/lib/withings/credentials";
|
||||
import { hasActivityScope, refreshAccessToken } from "@/lib/withings/client";
|
||||
import { hasActivityScope } from "@/lib/withings/client";
|
||||
import { obtainAccessToken } from "@/lib/integrations/oauth-kit/token";
|
||||
import { withingsTokenDescriptor } from "@/lib/withings/oauth-descriptor";
|
||||
import { isWithingsRefreshReauthFailure } from "@/lib/withings/sync";
|
||||
import { markReauthRequired } from "@/lib/integrations/status";
|
||||
|
||||
@@ -33,8 +33,6 @@ export const GET = apiHandler(async () => {
|
||||
where: { userId: user.id },
|
||||
select: {
|
||||
withingsUserId: true,
|
||||
accessToken: true,
|
||||
refreshToken: true,
|
||||
lastSyncedAt: true,
|
||||
tokenExpiresAt: true,
|
||||
createdAt: true,
|
||||
@@ -56,22 +54,14 @@ export const GET = apiHandler(async () => {
|
||||
const shouldRefresh = tokenExpiresAt.getTime() - 60_000 <= now;
|
||||
if (shouldRefresh) {
|
||||
try {
|
||||
const creds = await getUserWithingsCredentials(user.id);
|
||||
if (creds) {
|
||||
const refreshToken = decrypt(connection.refreshToken);
|
||||
const refreshed = await refreshAccessToken(refreshToken, creds);
|
||||
tokenExpiresAt = new Date(Date.now() + refreshed.expires_in * 1000);
|
||||
// The shared token lifecycle refreshes under the same lock the sync
|
||||
// takes, so a status read and a sync never spend the one-time refresh
|
||||
// token twice.
|
||||
const result = await obtainAccessToken(withingsTokenDescriptor, user.id);
|
||||
if (result.kind === "token" && result.expiresAt) {
|
||||
tokenExpiresAt = result.expiresAt;
|
||||
tokenExpired = tokenExpiresAt.getTime() <= Date.now();
|
||||
|
||||
await prisma.withingsConnection.update({
|
||||
where: { userId: user.id },
|
||||
data: {
|
||||
accessToken: encrypt(refreshed.access_token),
|
||||
refreshToken: encrypt(refreshed.refresh_token),
|
||||
tokenExpiresAt,
|
||||
},
|
||||
});
|
||||
} else if (tokenExpired) {
|
||||
} else if (result.kind !== "token" && tokenExpired) {
|
||||
tokenRefreshFailed = true;
|
||||
}
|
||||
} catch (error) {
|
||||
|
||||
@@ -126,14 +126,17 @@ beforeEach(() => {
|
||||
|
||||
describe("getValidToken — serialized rotating refresh", () => {
|
||||
it("re-reads token state under an advisory lock and reuses a peer winner", async () => {
|
||||
const stale = {
|
||||
id: "conn-1",
|
||||
withingsUserId: "wu-1",
|
||||
accessToken: "old-access",
|
||||
refreshToken: "old-refresh",
|
||||
tokenExpiresAt: new Date(Date.now() - 1000),
|
||||
};
|
||||
vi.mocked(prisma.withingsConnection.findUnique)
|
||||
.mockResolvedValueOnce({
|
||||
id: "conn-1",
|
||||
withingsUserId: "wu-1",
|
||||
accessToken: "old-access",
|
||||
refreshToken: "old-refresh",
|
||||
tokenExpiresAt: new Date(Date.now() - 1000),
|
||||
} as never)
|
||||
// The connection lookup, then the token read before the lock.
|
||||
.mockResolvedValueOnce(stale as never)
|
||||
.mockResolvedValueOnce(stale as never)
|
||||
.mockResolvedValueOnce({
|
||||
id: "conn-1",
|
||||
withingsUserId: "wu-1",
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
/**
|
||||
* Where a Withings grant lives and how its token endpoint is called, for the
|
||||
* shared token lifecycle (`src/lib/integrations/oauth-kit/token.ts`).
|
||||
*
|
||||
* Withings keeps an expiry on `WithingsConnection`, so the token is refreshed
|
||||
* ahead of it, and rotates the refresh token on every use.
|
||||
*/
|
||||
import type { OAuthTokenDescriptor } from "@/lib/integrations/oauth-kit/token";
|
||||
import { refreshAccessToken, type WithingsCredentials } from "./client";
|
||||
import { getUserWithingsCredentials } from "./credentials";
|
||||
|
||||
/** Refresh this long before the stored expiry. */
|
||||
export const WITHINGS_TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000;
|
||||
|
||||
export const withingsTokenDescriptor: OAuthTokenDescriptor<WithingsCredentials> =
|
||||
{
|
||||
provider: "withings",
|
||||
tracksExpiry: true,
|
||||
expiryBufferMs: WITHINGS_TOKEN_REFRESH_BUFFER_MS,
|
||||
async readGrant(db, userId) {
|
||||
const row = await db.withingsConnection.findUnique({
|
||||
where: { userId },
|
||||
select: { accessToken: true, refreshToken: true, tokenExpiresAt: true },
|
||||
});
|
||||
if (!row) return null;
|
||||
return {
|
||||
accessCiphertext: row.accessToken,
|
||||
refreshCiphertext: row.refreshToken,
|
||||
expiresAt: row.tokenExpiresAt,
|
||||
};
|
||||
},
|
||||
async writeGrant(db, userId, spent, next) {
|
||||
const { count } = await db.withingsConnection.updateMany({
|
||||
where: { userId, refreshToken: spent },
|
||||
data: {
|
||||
accessToken: next.accessCiphertext,
|
||||
refreshToken: next.refreshCiphertext,
|
||||
...(next.expiresAt ? { tokenExpiresAt: next.expiresAt } : {}),
|
||||
},
|
||||
});
|
||||
return count;
|
||||
},
|
||||
loadCredentials: (userId) => getUserWithingsCredentials(userId),
|
||||
refresh: (refreshToken, credentials) =>
|
||||
refreshAccessToken(refreshToken, credentials),
|
||||
};
|
||||
+21
-73
@@ -4,11 +4,9 @@
|
||||
*/
|
||||
import { prisma } from "@/lib/db";
|
||||
import type { MeasurementType, Prisma } from "@/generated/prisma/client";
|
||||
import { encrypt, decrypt } from "@/lib/crypto";
|
||||
import { getUnitForType } from "@/lib/validations/measurement";
|
||||
import {
|
||||
fetchMeasurements,
|
||||
refreshAccessToken,
|
||||
subscribeWebhook,
|
||||
WITHINGS_MEASTYPES,
|
||||
type WithingsMeasure,
|
||||
@@ -27,7 +25,6 @@ import {
|
||||
WITHINGS_PLAN_GATED_STATUS,
|
||||
type WithinPlanResult,
|
||||
} from "./api-plan";
|
||||
import { getUserWithingsCredentials } from "./credentials";
|
||||
import { annotate, getEvent } from "@/lib/logging/context";
|
||||
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
|
||||
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
|
||||
@@ -52,10 +49,8 @@ import {
|
||||
invalidateUserMeasurements,
|
||||
} from "@/lib/cache/invalidate";
|
||||
import type { SyncWriteResult } from "@/lib/outcome/written-outcome";
|
||||
import {
|
||||
acquireProviderTokenRefreshLock,
|
||||
PROVIDER_REFRESH_TRANSACTION_OPTIONS,
|
||||
} from "@/lib/integrations/oauth-refresh";
|
||||
import { obtainAccessToken } from "@/lib/integrations/oauth-kit/token";
|
||||
import { withingsTokenDescriptor } from "./oauth-descriptor";
|
||||
|
||||
/**
|
||||
* Build the callback URL handed to Withings at `Notify.subscribe` time.
|
||||
@@ -95,7 +90,9 @@ export interface GetValidWithingsTokenOptions {
|
||||
}
|
||||
|
||||
/**
|
||||
* Get valid access token for a user, refreshing if expired.
|
||||
* Get a valid access token for a user, refreshing it through the shared OAuth
|
||||
* token lifecycle (`src/lib/integrations/oauth-kit/token.ts`) when it is
|
||||
* within five minutes of its expiry.
|
||||
*/
|
||||
export async function getValidToken(
|
||||
userId: string,
|
||||
@@ -103,77 +100,28 @@ export async function getValidToken(
|
||||
): Promise<WithingsValidToken | null> {
|
||||
const connection = await prisma.withingsConnection.findUnique({
|
||||
where: { userId },
|
||||
select: { id: true, withingsUserId: true },
|
||||
});
|
||||
|
||||
if (!connection) return null;
|
||||
|
||||
if (connection.tokenExpiresAt.getTime() - 5 * 60 * 1000 >= Date.now()) {
|
||||
return {
|
||||
accessToken: decrypt(connection.accessToken),
|
||||
connection: {
|
||||
id: connection.id,
|
||||
withingsUserId: connection.withingsUserId,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const creds = await getUserWithingsCredentials(userId);
|
||||
if (!creds) {
|
||||
getEvent()?.addWarning(
|
||||
`No credentials found for user ${userId} during token refresh`,
|
||||
);
|
||||
await recordSyncFailure({
|
||||
userId,
|
||||
integration: "withings",
|
||||
kind: "reauth_required",
|
||||
message: "Withings credentials missing — token refresh skipped",
|
||||
errorCode: "credentials_missing",
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
return await prisma.$transaction(async (tx) => {
|
||||
await acquireProviderTokenRefreshLock(tx, "withings", userId);
|
||||
|
||||
const current = await tx.withingsConnection.findUnique({
|
||||
where: { userId },
|
||||
});
|
||||
if (!current) return null;
|
||||
|
||||
if (current.tokenExpiresAt.getTime() - 5 * 60 * 1000 >= Date.now()) {
|
||||
return {
|
||||
accessToken: decrypt(current.accessToken),
|
||||
connection: {
|
||||
id: current.id,
|
||||
withingsUserId: current.withingsUserId,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const newTokens = await refreshAccessToken(
|
||||
decrypt(current.refreshToken),
|
||||
creds,
|
||||
const result = await obtainAccessToken(withingsTokenDescriptor, userId);
|
||||
if (result.kind === "token") {
|
||||
return { accessToken: result.accessToken, connection };
|
||||
}
|
||||
if (result.kind === "no_credentials") {
|
||||
getEvent()?.addWarning(
|
||||
`No credentials found for user ${userId} during token refresh`,
|
||||
);
|
||||
const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000);
|
||||
|
||||
await tx.withingsConnection.update({
|
||||
where: { id: current.id },
|
||||
data: {
|
||||
accessToken: encrypt(newTokens.access_token),
|
||||
refreshToken: encrypt(newTokens.refresh_token),
|
||||
tokenExpiresAt: expiresAt,
|
||||
},
|
||||
await recordSyncFailure({
|
||||
userId,
|
||||
integration: "withings",
|
||||
kind: "reauth_required",
|
||||
message: "Withings credentials missing — token refresh skipped",
|
||||
errorCode: "credentials_missing",
|
||||
});
|
||||
|
||||
return {
|
||||
accessToken: newTokens.access_token,
|
||||
connection: {
|
||||
id: current.id,
|
||||
withingsUserId: current.withingsUserId,
|
||||
},
|
||||
};
|
||||
}, PROVIDER_REFRESH_TRANSACTION_OPTIONS);
|
||||
}
|
||||
return null;
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(`Token refresh failed for user ${userId}`);
|
||||
await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_MEASURES);
|
||||
|
||||
Reference in new issue
Block a user