mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 20:42:43 +00:00
Marc-André Bombeck
30aad51fac
Both hourly sync tails left the per-user analytics, correlations, targets, derived and achievements caches serving pre-sync bodies until their TTL lapsed — the correlations route documents the exact invariant these tails were breaking. Mirror the Fitbit tail: mark the cells stale whenever rows landed, on partial failures too.
869 lines
30 KiB
TypeScript
869 lines
30 KiB
TypeScript
/**
|
|
* Withings sync service — handles fetching and storing measurements,
|
|
* token refresh, and webhook subscription management.
|
|
*/
|
|
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,
|
|
} from "./client";
|
|
import { getUserWithingsCredentials } from "./credentials";
|
|
import { annotate, getEvent } from "@/lib/logging/context";
|
|
import { isP2002 } from "@/lib/prisma-errors";
|
|
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
|
|
import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
|
|
import {
|
|
isReauthRequired,
|
|
recordSyncFailure,
|
|
recordSyncSuccess,
|
|
type FailureKind,
|
|
} from "@/lib/integrations/status";
|
|
import {
|
|
classifyError,
|
|
WithingsApiError,
|
|
type WithingsClassification,
|
|
} from "./response-classifier";
|
|
import {
|
|
collapseToTypeDayKeys,
|
|
recomputeBucketsForMeasurement,
|
|
} from "@/lib/rollups/measurement-rollups";
|
|
import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
|
|
import {
|
|
invalidateUserDashboardSnapshot,
|
|
invalidateUserMeasurements,
|
|
} from "@/lib/cache/invalidate";
|
|
import type { SyncWriteResult } from "@/lib/outcome/written-outcome";
|
|
import {
|
|
acquireProviderTokenRefreshLock,
|
|
PROVIDER_REFRESH_TRANSACTION_OPTIONS,
|
|
} from "@/lib/integrations/oauth-refresh";
|
|
|
|
/**
|
|
* Build the callback URL handed to Withings at `Notify.subscribe` time.
|
|
*
|
|
* v1.4.25 W17a moved the shared secret from `?secret=…` (query
|
|
* parameter, captured by every reverse-proxy access log) to a path
|
|
* segment (`/api/withings/webhook/<secret>`). Withings has no
|
|
* mechanism for setting custom HTTP headers on outgoing notifications
|
|
* and never signs the body, so the callback URL is the only
|
|
* authenticity surface a subscriber controls — the path-segment form
|
|
* is the largest practical shift away from the loggable query string.
|
|
*
|
|
* When `WITHINGS_WEBHOOK_SECRET` is unset (dev / new install before
|
|
* provisioning) we fall back to the bare legacy URL so subscribe
|
|
* doesn't 500 — the route handler will then reject every inbound
|
|
* delivery with 401 anyway.
|
|
*/
|
|
export function getWithingsWebhookCallbackUrl(): string {
|
|
const baseUrl = `${process.env.NEXT_PUBLIC_APP_URL}/api/withings/webhook`;
|
|
const secret = process.env.WITHINGS_WEBHOOK_SECRET;
|
|
if (!secret) return baseUrl;
|
|
// Path-segment form. Withings preserves the full callback URL on
|
|
// every notification, so once a subscription is created with this
|
|
// URL every delivery carries the secret in the path rather than the
|
|
// query string. Encode for safety even though we expect a strong
|
|
// random secret.
|
|
return `${baseUrl}/${encodeURIComponent(secret)}`;
|
|
}
|
|
|
|
export interface WithingsValidToken {
|
|
accessToken: string;
|
|
connection: { id: string; withingsUserId: string };
|
|
}
|
|
|
|
export interface GetValidWithingsTokenOptions {
|
|
throwOnRefreshFailure?: boolean;
|
|
}
|
|
|
|
/**
|
|
* Get valid access token for a user, refreshing if expired.
|
|
*/
|
|
export async function getValidToken(
|
|
userId: string,
|
|
options: GetValidWithingsTokenOptions = {},
|
|
): Promise<WithingsValidToken | null> {
|
|
const connection = await prisma.withingsConnection.findUnique({
|
|
where: { userId },
|
|
});
|
|
|
|
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 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,
|
|
},
|
|
});
|
|
|
|
return {
|
|
accessToken: newTokens.access_token,
|
|
connection: {
|
|
id: current.id,
|
|
withingsUserId: current.withingsUserId,
|
|
},
|
|
};
|
|
}, PROVIDER_REFRESH_TRANSACTION_OPTIONS);
|
|
} catch (err) {
|
|
getEvent()?.addWarning(`Token refresh failed for user ${userId}`);
|
|
await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_MEASURES);
|
|
if (options.throwOnRefreshFailure) throw err;
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Incremental overlap window (ms). Withings backdates some readings (a scale
|
|
* step syncs to the cloud minutes after the reading's timestamp), and
|
|
* `lastSyncedAt` advances on every success — including a healthy 200-with-0
|
|
* cycle (F-SYNC-1). A too-tight overlap let a reading that landed just before
|
|
* the next cycle's `now()` slip through the gap. 10 minutes is comfortably wider
|
|
* than the observed backdating skew while staying far short of a full re-scan;
|
|
* the upserts are idempotent, so a wider overlap only re-touches a handful of
|
|
* rows. (Still narrower than Google's 24h / Oura's 7-day because Withings is
|
|
* webhook-primary — this overlap is the poll-catch-up safety net, not the main
|
|
* path.)
|
|
*/
|
|
export const WITHINGS_INCREMENTAL_OVERLAP_MS = 10 * 60 * 1000;
|
|
|
|
/**
|
|
* Sync measurements from Withings for a given user.
|
|
* Fetches data since last sync (or last 30 days if first sync).
|
|
*
|
|
* Returns what was written AND whether any part of the run did not land, so
|
|
* the caller can tell a quiet week apart from a run that refused every row.
|
|
*
|
|
* Status-tracking contract:
|
|
* - On any failure (refresh failure, fetch failure, downstream
|
|
* measurement-upsert that fails ALL items) we call
|
|
* `recordSyncFailure` so the IntegrationStatus row + audit log
|
|
* reflect the burst, and so the persistent-failure threshold can
|
|
* trip an admin alert.
|
|
* - On full success we call `recordSyncSuccess` to clear the streak.
|
|
* - If the connection is parked at `error_reauth` we short-circuit
|
|
* immediately — the user has to reconnect first.
|
|
*/
|
|
export async function syncUserMeasurements(
|
|
userId: string,
|
|
opts: { fullSync?: boolean } = {},
|
|
): Promise<SyncWriteResult> {
|
|
// Park: if the last burst said reauth_required, do nothing until
|
|
// the OAuth callback flips state back to "connected". The run wrote
|
|
// nothing AND did not complete, so it reports as a failure — a bare 0
|
|
// read as "nothing new", which is how a parked connection could keep
|
|
// answering with a green tick.
|
|
if (await isReauthRequired(userId, "withings")) {
|
|
getEvent()?.addWarning(
|
|
`withings sync skipped for ${userId}: parked at error_reauth`,
|
|
);
|
|
return { imported: 0, failed: true };
|
|
}
|
|
|
|
const tokenInfo = await getValidToken(userId);
|
|
// Note: getValidToken already calls recordSyncFailure on the refresh
|
|
// path, so we don't double-record here.
|
|
if (!tokenInfo) return { imported: 0, failed: true };
|
|
|
|
const connection = await prisma.withingsConnection.findUnique({
|
|
where: { userId },
|
|
});
|
|
// No connection row is not a failure — there is nothing to sync.
|
|
if (!connection) return { imported: 0, failed: false };
|
|
|
|
const startDate = opts.fullSync
|
|
? undefined
|
|
: connection.lastSyncedAt
|
|
? new Date(
|
|
connection.lastSyncedAt.getTime() - WITHINGS_INCREMENTAL_OVERLAP_MS,
|
|
)
|
|
: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); // 30 days back
|
|
|
|
let fetched: Awaited<ReturnType<typeof fetchMeasurements>>;
|
|
try {
|
|
fetched = await fetchMeasurements(tokenInfo.accessToken, startDate);
|
|
} catch (err) {
|
|
await recordWithingsSyncFailure(userId, err, WITHINGS_LEG_MEASURES);
|
|
throw err;
|
|
}
|
|
|
|
// The app's own plausibility band applies to a provider reading exactly as
|
|
// it applies to a typed one. A remote that reports a pulse in the millions
|
|
// is reporting a decode slip, not a heartbeat, and the row would poison
|
|
// every mean and personal band derived from it. Refuse it here, before the
|
|
// update branch below can overwrite a good stored value with it.
|
|
const measures = dropImplausibleMeasurements("withings", fetched, (m) => ({
|
|
type: m.type,
|
|
value: m.value,
|
|
}));
|
|
|
|
let imported = 0;
|
|
// v1.4.39.1 — track every (type, measuredAt) we touched so the
|
|
// persistent rollup tier can be re-folded at the end of the sync.
|
|
// Pre-fix, Withings-ingested BP / weight / pulse / body-fat rows
|
|
// skipped the rollup write hook entirely: subsequent dashboard chart
|
|
// fetches with `source=rollup` saw fewer DAY buckets than the live
|
|
// measurements table, painting the "Noch nicht genug Daten" empty
|
|
// state for the 30-day range on accounts whose recent data only came
|
|
// through Withings. We collect first, fold once at the end so a 100-
|
|
// row monthly catch-up costs at most ~N (type, day) recomputes rather
|
|
// than N per-row hooks. Best-effort: a populator hiccup never fails
|
|
// the user's sync.
|
|
const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
|
|
const insertedArrivals: Array<{
|
|
id: string;
|
|
type: MeasurementType;
|
|
measuredAt: Date;
|
|
}> = [];
|
|
|
|
// v1.28.39 — hold-watermark-on-hard-failure (mirrors google-health /
|
|
// fitbit's `hardFailStorage` verdict). A per-row write that HARD-fails
|
|
// (anything but a benign P2002 idempotent-write collision) must not let
|
|
// `lastSyncedAt` advance past the unpersisted reading — the 10-min overlap
|
|
// window would otherwise never re-cover it and the reading is stranded
|
|
// forever. Track any hard failure so the watermark stamp + `recordSyncSuccess`
|
|
// below are held and the failure is recorded for the next tick to retry.
|
|
let anyRowFailed = false;
|
|
let firstRowError: unknown = null;
|
|
|
|
for (const m of measures) {
|
|
const measType = m.type as MeasurementType;
|
|
try {
|
|
// v1.4.25 W17b/c — Migration 0055 added `sleepStage` to the
|
|
// composite unique with `NULLS NOT DISTINCT`, so a non-sleep
|
|
// upsert keyed on (userId, type, measuredAt, source) with
|
|
// sleepStage IS NULL is still safe. But Prisma's typed compound
|
|
// input requires a non-null `sleepStage`, so we model the
|
|
// idempotent write as a `findFirst` + `create`/`update`. The
|
|
// unique index still serializes concurrent inserts, so the worst
|
|
// case is a Prisma P2002 we catch and ignore.
|
|
const existing = await prisma.measurement.findFirst({
|
|
where: {
|
|
userId,
|
|
type: measType,
|
|
measuredAt: m.measuredAt,
|
|
source: "WITHINGS",
|
|
sleepStage: null,
|
|
},
|
|
select: { id: true },
|
|
});
|
|
if (existing) {
|
|
await prisma.measurement.update({
|
|
where: { id: existing.id },
|
|
// `deletedAt: null` is a no-op on a live row, a deliberate
|
|
// RESURRECTION on a tombstoned one — Withings is the source of
|
|
// truth for its own rows, so a re-fetched measure brings the row
|
|
// back (mirrors Google / Fitbit).
|
|
data: { value: m.value, deletedAt: null },
|
|
});
|
|
} else {
|
|
const created = await prisma.measurement.create({
|
|
data: {
|
|
userId,
|
|
type: measType,
|
|
value: m.value,
|
|
unit: getUnitForType(m.type),
|
|
measuredAt: m.measuredAt,
|
|
source: "WITHINGS",
|
|
},
|
|
select: { id: true, type: true, measuredAt: true },
|
|
});
|
|
insertedArrivals.push(created);
|
|
}
|
|
touched.push({ type: measType, measuredAt: m.measuredAt });
|
|
imported++;
|
|
} catch (err) {
|
|
if (isP2002(err)) {
|
|
// Benign: the composite unique index serialised a concurrent insert of
|
|
// this exact reading — the row is present, nothing is lost. Warn and
|
|
// continue WITHOUT holding the watermark (this is the expected
|
|
// idempotent-write race the findFirst+create models around).
|
|
getEvent()?.addWarning(
|
|
`Withings measure upsert hit an idempotent P2002 (row already present): ${err}`,
|
|
);
|
|
} else {
|
|
// A genuine write failure strands this reading. Flag the cycle so the
|
|
// watermark is held and the next scheduled tick refetches the overlap
|
|
// window and retries.
|
|
anyRowFailed = true;
|
|
if (firstRowError === null) firstRowError = err;
|
|
getEvent()?.addWarning(`Failed to upsert measure: ${err}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
void emitInsertedMeasurementArrivals(
|
|
userId,
|
|
insertedArrivals,
|
|
"withings",
|
|
).catch(() => {});
|
|
|
|
// v1.4.39.1 — refresh the persistent rollup table for every distinct
|
|
// (type, day) the sync touched. The chart-data + analytics read paths
|
|
// consume these buckets via `source=rollup`; without this hook,
|
|
// Withings-only metrics stayed off the rollup tier and the dashboard
|
|
// chart at the 30-day range under-counted recent days.
|
|
try {
|
|
const keys = collapseToTypeDayKeys(touched);
|
|
for (const k of keys) {
|
|
await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
|
|
}
|
|
|
|
// v1.8.0 — a Withings sync that lands new weight / BP / pulse /
|
|
// body-comp rows dirties the matching per-metric assessment caches.
|
|
// Drop them so the next mount / nightly warm pass regenerates
|
|
// against the new data. Fire-and-forget: never fails the sync.
|
|
invalidateStatusInsightsForTypes(
|
|
userId,
|
|
keys.map((k) => k.type),
|
|
).catch((err) => {
|
|
getEvent()?.addWarning(
|
|
`withings: status-insight invalidate failed for ${userId}: ${err}`,
|
|
);
|
|
});
|
|
|
|
// v1.18.9 (#38) — hard-evict the dashboard snapshot so a focus-refetch
|
|
// after this server-side sync returns the freshly-imported readings.
|
|
// The server-side Withings path produces no client mutation event, so
|
|
// without this the snapshot's `cachedSwr` entry kept serving the
|
|
// pre-sync body until its ~180 s TTL lapsed.
|
|
if (keys.length > 0) {
|
|
invalidateUserDashboardSnapshot(userId);
|
|
// Background-sync posture (mirrors the Fitbit tail): mark the
|
|
// per-user analytics / correlations / targets / achievements cells
|
|
// stale so the imported rows reach the cached readers before their
|
|
// TTL lapses — the correlations route documents this invariant.
|
|
invalidateUserMeasurements(userId);
|
|
}
|
|
} catch (err) {
|
|
getEvent()?.addWarning(
|
|
`withings: rollup recompute failed for ${userId}: ${err}`,
|
|
);
|
|
}
|
|
|
|
// v1.28.39 — HOLD the watermark on any hard row failure. Implements the
|
|
// documented status-tracking contract above ("downstream measurement-upsert
|
|
// that fails ALL items → recordSyncFailure"), extended to a partial hard
|
|
// failure: a single hard-failed row must not advance `lastSyncedAt` past the
|
|
// reading it failed to persist, or the 10-min overlap window loses it. Record
|
|
// the failure (classified transient by default → the connection stays live
|
|
// and the next tick retries from the un-advanced watermark) and return
|
|
// WITHOUT stamping the watermark or success. The rows that DID write kept
|
|
// their rollup recompute above.
|
|
if (anyRowFailed) {
|
|
annotate({
|
|
action: {
|
|
name: "withings.sync.watermark_held",
|
|
details: { imported },
|
|
},
|
|
});
|
|
await recordWithingsSyncFailure(
|
|
userId,
|
|
firstRowError ?? new Error("Withings measure row write failed"),
|
|
WITHINGS_LEG_MEASURES,
|
|
);
|
|
return { imported, failed: true };
|
|
}
|
|
|
|
// Update last synced timestamp
|
|
await prisma.withingsConnection.update({
|
|
where: { userId },
|
|
data: { lastSyncedAt: new Date() },
|
|
});
|
|
|
|
// Record success — even when measures.length === 0, completing the
|
|
// round-trip is the signal "the connection is healthy". A user who
|
|
// doesn't weigh themselves for a week shouldn't see "error" on
|
|
// their settings page.
|
|
await recordSyncSuccess(userId, "withings", {
|
|
leg: WITHINGS_LEG_MEASURES,
|
|
});
|
|
|
|
return { imported, failed: false };
|
|
}
|
|
|
|
/**
|
|
* Map a Withings response classification onto the `FailureKind` carried
|
|
* into `recordSyncFailure`. Pure function — no IO, no state.
|
|
*
|
|
* `success` → not a failure; callers must not invoke this.
|
|
* `transient` → `transient` (retry next sync)
|
|
* `reauth_required` → `reauth_required` (park at `error_reauth`)
|
|
* `persistent` → `persistent` (audited; next sync still runs)
|
|
*/
|
|
/**
|
|
* v1.4.42 — shared failure-recording shape for the two Withings sync
|
|
* catch-blocks. The typed `WithingsApiError` carries the classification
|
|
* verdict directly; `classifyError`'s regex fallback handles plain
|
|
* `Error` instances (e.g. a pg-boss job retry that lost the prototype
|
|
* during the JSON round-trip).
|
|
*/
|
|
/**
|
|
* The four Withings sync legs. Each runs on its own cron at its own minute and
|
|
* writes the SAME `(user, withings)` ledger row, so a failure has to say which
|
|
* leg it came from — otherwise the next sibling to succeed clears an error it
|
|
* did not cause, which is what kept a broken leg invisible.
|
|
*/
|
|
export type WithingsSyncLeg = "measures" | "activity" | "sleep" | "ecg";
|
|
|
|
export const WITHINGS_LEG_MEASURES: WithingsSyncLeg = "measures";
|
|
export const WITHINGS_LEG_ACTIVITY: WithingsSyncLeg = "activity";
|
|
export const WITHINGS_LEG_SLEEP: WithingsSyncLeg = "sleep";
|
|
export const WITHINGS_LEG_ECG: WithingsSyncLeg = "ecg";
|
|
|
|
export async function recordWithingsSyncFailure(
|
|
userId: string,
|
|
err: unknown,
|
|
leg: WithingsSyncLeg,
|
|
): Promise<void> {
|
|
const message = err instanceof Error ? err.message : String(err);
|
|
await recordSyncFailure({
|
|
userId,
|
|
integration: "withings",
|
|
kind: classificationToFailureKind(classifyError(err)),
|
|
message,
|
|
leg,
|
|
errorCode:
|
|
err instanceof WithingsApiError
|
|
? err.withingsStatus?.toString()
|
|
: extractWithingsStatus(message),
|
|
});
|
|
}
|
|
|
|
export function classificationToFailureKind(
|
|
classification: WithingsClassification,
|
|
): FailureKind {
|
|
switch (classification) {
|
|
case "reauth_required":
|
|
return "reauth_required";
|
|
case "persistent":
|
|
return "persistent";
|
|
case "transient":
|
|
return "transient";
|
|
case "success":
|
|
// Defensive: a caller asking for the FailureKind of a success
|
|
// is a contract bug. Default to `transient` so the audit log
|
|
// still surfaces the bug rather than silently ignoring it.
|
|
return "transient";
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Withings status codes that indicate a permanent-revoke condition
|
|
* (the refresh_token will not work again — the user has to redo OAuth).
|
|
*
|
|
* Documented values per
|
|
* https://developer.withings.com/api-reference#tag/oauth2/operation/oauth2-getaccesstoken :
|
|
*
|
|
* 100 → "Authentication failed"
|
|
* 101 → "Invalid token"
|
|
* 102 → "User does not exist"
|
|
* 200..299 → various OAuth/grant failures (treated as reauth-required
|
|
* by Withings' own libraries)
|
|
*
|
|
* The error message format from `refreshAccessToken()` is
|
|
* "Withings refresh error: <status> - <error>".
|
|
*
|
|
* v1.4.42 W6 — superseded for new code paths by
|
|
* `classifyError`/`classifyWithingsResponse` in `response-classifier.ts`.
|
|
* Kept for the W17b/c sync-activity + sync-sleep catch-blocks that
|
|
* haven't yet been migrated; both paths now also benefit from the
|
|
* typed-error throw upstream because `classifyError` falls back to the
|
|
* same regex.
|
|
*/
|
|
export function isWithingsRefreshReauthFailure(message: string): boolean {
|
|
const status = extractWithingsStatus(message);
|
|
if (!status) return false;
|
|
const n = Number.parseInt(status, 10);
|
|
if (!Number.isFinite(n)) return false;
|
|
if (n === 100 || n === 101 || n === 102) return true;
|
|
if (n >= 200 && n <= 299) return true;
|
|
return false;
|
|
}
|
|
|
|
export function extractWithingsStatus(message: string): string | undefined {
|
|
// Captures the digit run after "Withings <verb> error:" — same shape
|
|
// exchangeCode and refreshAccessToken use.
|
|
const m = /Withings\s+\w+\s+error:\s*(\d+)/.exec(message);
|
|
return m?.[1];
|
|
}
|
|
|
|
/**
|
|
* Withings notify appli categories HealthLog ingests. Each category requires
|
|
* an independent upstream subscription.
|
|
*/
|
|
export const WITHINGS_NOTIFY_APPLIS = [1, 2, 4, 16, 44] as const;
|
|
|
|
export type WithingsNotifyAppli = (typeof WITHINGS_NOTIFY_APPLIS)[number];
|
|
export type WithingsNotifyAppliKey = `${WithingsNotifyAppli}`;
|
|
export type WithingsWebhookSubscriptionStatus =
|
|
"pending" | "success" | "transient" | "persistent" | "reauth_required";
|
|
|
|
export interface WithingsWebhookSubscriptionOutcome {
|
|
status: WithingsWebhookSubscriptionStatus;
|
|
attemptCount: number;
|
|
lastAttemptAt: string | null;
|
|
nextRetryAt: string | null;
|
|
withingsStatus: number | null;
|
|
}
|
|
|
|
export interface WithingsWebhookSubscriptionStateV1 {
|
|
version: 1;
|
|
outcomes: Record<WithingsNotifyAppliKey, WithingsWebhookSubscriptionOutcome>;
|
|
}
|
|
|
|
export const WITHINGS_SUBSCRIPTION_BASE_RETRY_MS = 5 * 60 * 1000;
|
|
export const WITHINGS_SUBSCRIPTION_MAX_RETRY_MS = 24 * 60 * 60 * 1000;
|
|
|
|
const WITHINGS_SUBSCRIPTION_STATUSES: Record<
|
|
WithingsWebhookSubscriptionStatus,
|
|
true
|
|
> = {
|
|
pending: true,
|
|
success: true,
|
|
transient: true,
|
|
persistent: true,
|
|
reauth_required: true,
|
|
};
|
|
|
|
function createPendingSubscriptionOutcome(
|
|
now: Date,
|
|
): WithingsWebhookSubscriptionOutcome {
|
|
return {
|
|
status: "pending",
|
|
attemptCount: 0,
|
|
lastAttemptAt: null,
|
|
nextRetryAt: now.toISOString(),
|
|
withingsStatus: null,
|
|
};
|
|
}
|
|
|
|
export function createPendingWithingsWebhookSubscriptionState(
|
|
now: Date = new Date(),
|
|
): WithingsWebhookSubscriptionStateV1 {
|
|
return {
|
|
version: 1,
|
|
outcomes: Object.fromEntries(
|
|
WITHINGS_NOTIFY_APPLIS.map((appli) => [
|
|
String(appli),
|
|
createPendingSubscriptionOutcome(now),
|
|
]),
|
|
) as WithingsWebhookSubscriptionStateV1["outcomes"],
|
|
};
|
|
}
|
|
|
|
function isObject(value: unknown): value is Record<string, unknown> {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
|
}
|
|
|
|
function isDateStringOrNull(value: unknown): value is string | null {
|
|
return (
|
|
value === null ||
|
|
(typeof value === "string" && Number.isFinite(Date.parse(value)))
|
|
);
|
|
}
|
|
|
|
function parseSubscriptionOutcome(
|
|
value: unknown,
|
|
): WithingsWebhookSubscriptionOutcome | null {
|
|
if (!isObject(value)) return null;
|
|
if (
|
|
typeof value.status !== "string" ||
|
|
!WITHINGS_SUBSCRIPTION_STATUSES[
|
|
value.status as WithingsWebhookSubscriptionStatus
|
|
] ||
|
|
!Number.isSafeInteger(value.attemptCount) ||
|
|
(value.attemptCount as number) < 0 ||
|
|
!isDateStringOrNull(value.lastAttemptAt) ||
|
|
!isDateStringOrNull(value.nextRetryAt) ||
|
|
!(
|
|
value.withingsStatus === null ||
|
|
(typeof value.withingsStatus === "number" &&
|
|
Number.isSafeInteger(value.withingsStatus))
|
|
)
|
|
) {
|
|
return null;
|
|
}
|
|
|
|
const status = value.status as WithingsWebhookSubscriptionStatus;
|
|
if (
|
|
(status === "transient" || status === "pending") &&
|
|
value.nextRetryAt === null
|
|
) {
|
|
return null;
|
|
}
|
|
if (
|
|
status !== "transient" &&
|
|
status !== "pending" &&
|
|
value.nextRetryAt !== null
|
|
) {
|
|
return null;
|
|
}
|
|
|
|
return {
|
|
status,
|
|
attemptCount: value.attemptCount as number,
|
|
lastAttemptAt: value.lastAttemptAt,
|
|
nextRetryAt: value.nextRetryAt,
|
|
withingsStatus: value.withingsStatus as number | null,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Persisted JSON is an untrusted compatibility boundary. Unknown versions or
|
|
* malformed outcomes reset to a fully pending v1 state rather than silently
|
|
* treating a category as subscribed.
|
|
*/
|
|
export function parseWithingsWebhookSubscriptionState(
|
|
value: unknown,
|
|
now: Date = new Date(),
|
|
): WithingsWebhookSubscriptionStateV1 {
|
|
const pending = createPendingWithingsWebhookSubscriptionState(now);
|
|
if (!isObject(value) || value.version !== 1 || !isObject(value.outcomes)) {
|
|
return pending;
|
|
}
|
|
|
|
const outcomes = { ...pending.outcomes };
|
|
for (const appli of WITHINGS_NOTIFY_APPLIS) {
|
|
const key = String(appli) as WithingsNotifyAppliKey;
|
|
const raw = value.outcomes[key];
|
|
if (raw === undefined) continue;
|
|
const parsed = parseSubscriptionOutcome(raw);
|
|
if (!parsed) return pending;
|
|
outcomes[key] = parsed;
|
|
}
|
|
return { version: 1, outcomes };
|
|
}
|
|
|
|
function subscriptionBackoffMs(attemptCount: number): number {
|
|
const exponent = Math.min(Math.max(attemptCount - 1, 0), 30);
|
|
return Math.min(
|
|
WITHINGS_SUBSCRIPTION_BASE_RETRY_MS * 2 ** exponent,
|
|
WITHINGS_SUBSCRIPTION_MAX_RETRY_MS,
|
|
);
|
|
}
|
|
|
|
function earliestSubscriptionRetryAt(
|
|
state: WithingsWebhookSubscriptionStateV1,
|
|
): Date | null {
|
|
let earliest: number | null = null;
|
|
for (const outcome of Object.values(state.outcomes)) {
|
|
if (
|
|
(outcome.status !== "pending" && outcome.status !== "transient") ||
|
|
outcome.nextRetryAt === null
|
|
) {
|
|
continue;
|
|
}
|
|
const timestamp = Date.parse(outcome.nextRetryAt);
|
|
if (earliest === null || timestamp < earliest) earliest = timestamp;
|
|
}
|
|
return earliest === null ? null : new Date(earliest);
|
|
}
|
|
|
|
async function persistWithingsWebhookSubscriptionState(
|
|
connectionId: string,
|
|
state: WithingsWebhookSubscriptionStateV1,
|
|
): Promise<void> {
|
|
await prisma.withingsConnection.update({
|
|
where: { id: connectionId },
|
|
data: {
|
|
webhookSubscriptionState: state as unknown as Prisma.InputJsonValue,
|
|
webhookSubscriptionRetryAt: earliestSubscriptionRetryAt(state),
|
|
},
|
|
});
|
|
}
|
|
|
|
function subscriptionFailureOutcome(
|
|
previous: WithingsWebhookSubscriptionOutcome,
|
|
err: unknown,
|
|
now: Date,
|
|
): WithingsWebhookSubscriptionOutcome {
|
|
const classified = classifyError(err);
|
|
const classification = classified === "success" ? "transient" : classified;
|
|
const attemptCount = previous.attemptCount + 1;
|
|
const withingsStatus =
|
|
err instanceof WithingsApiError &&
|
|
typeof err.withingsStatus === "number" &&
|
|
Number.isSafeInteger(err.withingsStatus)
|
|
? err.withingsStatus
|
|
: null;
|
|
return {
|
|
status: classification,
|
|
attemptCount,
|
|
lastAttemptAt: now.toISOString(),
|
|
nextRetryAt:
|
|
classification === "transient"
|
|
? new Date(
|
|
now.getTime() + subscriptionBackoffMs(attemptCount),
|
|
).toISOString()
|
|
: null,
|
|
withingsStatus,
|
|
};
|
|
}
|
|
|
|
async function reconcileWithingsWebhookSubscriptions(
|
|
userId: string,
|
|
now: Date,
|
|
): Promise<void> {
|
|
const connection = await prisma.withingsConnection.findUnique({
|
|
where: { userId },
|
|
select: {
|
|
id: true,
|
|
webhookSubscriptionState: true,
|
|
},
|
|
});
|
|
if (!connection) return;
|
|
|
|
const state = parseWithingsWebhookSubscriptionState(
|
|
connection.webhookSubscriptionState,
|
|
now,
|
|
);
|
|
const dueApplis = WITHINGS_NOTIFY_APPLIS.filter((appli) => {
|
|
const outcome = state.outcomes[String(appli) as WithingsNotifyAppliKey];
|
|
return (
|
|
(outcome.status === "pending" || outcome.status === "transient") &&
|
|
outcome.nextRetryAt !== null &&
|
|
Date.parse(outcome.nextRetryAt) <= now.getTime()
|
|
);
|
|
});
|
|
if (dueApplis.length === 0) return;
|
|
|
|
let tokenInfo: WithingsValidToken | null;
|
|
try {
|
|
tokenInfo = await getValidToken(userId, { throwOnRefreshFailure: true });
|
|
} catch (err) {
|
|
for (const appli of dueApplis) {
|
|
const key = String(appli) as WithingsNotifyAppliKey;
|
|
state.outcomes[key] = subscriptionFailureOutcome(
|
|
state.outcomes[key],
|
|
err,
|
|
now,
|
|
);
|
|
await persistWithingsWebhookSubscriptionState(connection.id, state);
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (!tokenInfo) {
|
|
for (const appli of dueApplis) {
|
|
const key = String(appli) as WithingsNotifyAppliKey;
|
|
const previous = state.outcomes[key];
|
|
state.outcomes[key] = {
|
|
status: "reauth_required",
|
|
attemptCount: previous.attemptCount + 1,
|
|
lastAttemptAt: now.toISOString(),
|
|
nextRetryAt: null,
|
|
withingsStatus: null,
|
|
};
|
|
await persistWithingsWebhookSubscriptionState(connection.id, state);
|
|
}
|
|
return;
|
|
}
|
|
|
|
const callbackUrl = getWithingsWebhookCallbackUrl();
|
|
for (const appli of dueApplis) {
|
|
const key = String(appli) as WithingsNotifyAppliKey;
|
|
const previous = state.outcomes[key];
|
|
try {
|
|
await subscribeWebhook(tokenInfo.accessToken, callbackUrl, appli);
|
|
state.outcomes[key] = {
|
|
status: "success",
|
|
attemptCount: previous.attemptCount + 1,
|
|
lastAttemptAt: now.toISOString(),
|
|
nextRetryAt: null,
|
|
withingsStatus: null,
|
|
};
|
|
getEvent()?.addMeta(`webhook_subscribed_${appli}`, userId);
|
|
} catch (err) {
|
|
state.outcomes[key] = subscriptionFailureOutcome(previous, err, now);
|
|
const outcome = state.outcomes[key];
|
|
getEvent()?.addWarning(
|
|
`Webhook subscription failed (appli=${appli}, classification=${outcome.status}, status=${outcome.withingsStatus ?? "unknown"})`,
|
|
);
|
|
}
|
|
await persistWithingsWebhookSubscriptionState(connection.id, state);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Initial setup and reconnect both reconcile the durable state. Successful
|
|
* categories are retained; reconnect resets the JSON before calling this.
|
|
*/
|
|
export async function setupWebhook(
|
|
userId: string,
|
|
now: Date = new Date(),
|
|
): Promise<void> {
|
|
await reconcileWithingsWebhookSubscriptions(userId, now);
|
|
}
|
|
|
|
/**
|
|
* Hourly repair discovers only rows whose indexed retry timestamp is due, then
|
|
* narrows again to due pending/transient categories inside the validated JSON.
|
|
*/
|
|
export async function retryDueWithingsWebhookSubscriptions(
|
|
now: Date = new Date(),
|
|
): Promise<number> {
|
|
const connections = await prisma.withingsConnection.findMany({
|
|
where: { webhookSubscriptionRetryAt: { lte: now } },
|
|
select: { userId: true },
|
|
});
|
|
for (const connection of connections) {
|
|
await reconcileWithingsWebhookSubscriptions(connection.userId, now);
|
|
}
|
|
return connections.length;
|
|
}
|