mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
feat(google-health): wire the OAuth routes, sync jobs, and source priority
Add the request surface and the background sync for the Google Health provider, all built on the shared integration plumbing. - Routes under /api/google-health: connect (mint the PKCE pair and state nonce, redirect to Google), callback (timing-safe CSRF, atomic state consume, token exchange, connection upsert, clear the reauth flag, enqueue backfill), credentials (encrypt the operator's BYO client id and secret field-by-field), disconnect, status, and a rate-limited manual sync. - Jobs: a boot-time backfill, an hourly poll cohort, and a daily oauth-state cleanup, registered under the existing dead-queue contract. - Rank GOOGLE_HEALTH just after Fitbit in the per-metric source ladders; it is the successor API for the same devices. - Add GOOGLE_HEALTH to the readable measurement-source enum but leave it out of the writable set — the provider is server-owned, so a client cannot attribute a write to it. - Register the two new encrypted token columns and the two BYO-client columns in the encryption-rotation registry and the rotation script.
This commit is contained in:
1 parent
fb17a4e9c6
commit
b26a18d35e
15 files changed
+1107
-10
No files matched your search
@@ -178,6 +178,8 @@ async function main() {
|
||||
"whoopClientSecretEncrypted",
|
||||
"fitbitClientIdEncrypted",
|
||||
"fitbitClientSecretEncrypted",
|
||||
"googleHealthClientIdEncrypted",
|
||||
"googleHealthClientSecretEncrypted",
|
||||
"nightscoutUrlEncrypted",
|
||||
"nightscoutTokenEncrypted",
|
||||
"polarAccessTokenEncrypted",
|
||||
@@ -223,6 +225,13 @@ async function main() {
|
||||
prisma.fitbitConnection,
|
||||
),
|
||||
);
|
||||
results.push(
|
||||
await rotateStringColumn(
|
||||
"GoogleHealthConnection",
|
||||
field,
|
||||
prisma.googleHealthConnection,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
// ───── AppSettings — operator credentials (String) ─────
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
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,
|
||||
resolveGoogleHealthUserId,
|
||||
getGoogleHealthScopeString,
|
||||
} from "@/lib/google-health/client";
|
||||
import { getUserGoogleHealthCredentials } from "@/lib/google-health/credentials";
|
||||
import { GOOGLE_HEALTH_OAUTH_STATE_COOKIE } from "@/lib/google-health/oauth-state";
|
||||
import { GOOGLE_HEALTH_BACKFILL_QUEUE } from "@/lib/jobs/google-health-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 the Google Health integration (v1.26.0).
|
||||
* Mirrors the Fitbit callback with PKCE: the `state` param is a random
|
||||
* base64url nonce keyed against the `GoogleHealthOAuthState` ledger, which also
|
||||
* carries the PKCE `code_verifier`. 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 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 with the stored verifier, fetch the Google
|
||||
* Health profile for the external `googleUserId`, persist the encrypted
|
||||
* `GoogleHealthConnection`, clear any prior reauth state (`needsReauth:false` +
|
||||
* the status ledger), and enqueue the self-converging history backfill.
|
||||
*/
|
||||
const ERR = (reason: string) =>
|
||||
NextResponse.redirect(
|
||||
new URL(
|
||||
`/settings/integrations?googleHealth=error&reason=${reason}`,
|
||||
process.env.NEXT_PUBLIC_APP_URL!,
|
||||
),
|
||||
);
|
||||
|
||||
export const GET = apiHandler(async (request: NextRequest) => {
|
||||
const { user } = await requireAuth();
|
||||
annotate({ action: { name: "google_health.callback" } });
|
||||
|
||||
const { searchParams } = new URL(request.url);
|
||||
const code = searchParams.get("code");
|
||||
const state = searchParams.get("state");
|
||||
const storedState = request.cookies.get(
|
||||
GOOGLE_HEALTH_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 (and the PKCE verifier read)
|
||||
// run against the consumed payload. P2025 means the nonce was already
|
||||
// consumed (replay) or never existed — benign. 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;
|
||||
codeVerifier: string | null;
|
||||
} | null = null;
|
||||
try {
|
||||
stateRow = await prisma.googleHealthOAuthState.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 (!stateRow.codeVerifier) {
|
||||
annotate({ meta: { reason: "no_verifier" } });
|
||||
return ERR("no_verifier");
|
||||
}
|
||||
|
||||
if (!code) {
|
||||
return ERR("nocode");
|
||||
}
|
||||
|
||||
try {
|
||||
const creds = await getUserGoogleHealthCredentials(user.id);
|
||||
if (!creds) {
|
||||
return ERR("nocreds");
|
||||
}
|
||||
|
||||
const tokens = await exchangeCode(code, stateRow.codeVerifier, creds);
|
||||
|
||||
// The initial code exchange MUST return a refresh token (the connect route
|
||||
// forces `prompt=consent` + `access_type=offline`). If Google ever omits
|
||||
// it, refuse to persist — a stored empty refresh token silently bricks
|
||||
// every future token refresh (the refresh POST then 400s and the
|
||||
// connection dies the moment the access token expires). Error out cleanly
|
||||
// so the user re-consents rather than landing a dead connection.
|
||||
if (!tokens.refresh_token) {
|
||||
getEvent()?.addWarning(
|
||||
`google-health callback: token response carried no refresh_token for user ${user.id}`,
|
||||
);
|
||||
annotate({ meta: { reason: "no_refresh_token" } });
|
||||
return ERR("no_refresh_token");
|
||||
}
|
||||
|
||||
const profile = await fetchProfile(tokens.access_token);
|
||||
const googleUserId = resolveGoogleHealthUserId(profile);
|
||||
const expiresAt = new Date(Date.now() + tokens.expires_in * 1000);
|
||||
const scope = tokens.scope ?? getGoogleHealthScopeString();
|
||||
|
||||
await prisma.googleHealthConnection.upsert({
|
||||
where: { userId: user.id },
|
||||
update: {
|
||||
googleUserId,
|
||||
accessToken: encrypt(tokens.access_token),
|
||||
refreshToken: encrypt(tokens.refresh_token),
|
||||
tokenExpiresAt: expiresAt,
|
||||
scope,
|
||||
// Fresh consent clears the soft-disconnect flag so the card drops the
|
||||
// reconnect CTA and the sync path stops short-circuiting.
|
||||
needsReauth: false,
|
||||
backfillCompletedAt: null,
|
||||
},
|
||||
create: {
|
||||
userId: user.id,
|
||||
googleUserId,
|
||||
accessToken: encrypt(tokens.access_token),
|
||||
refreshToken: encrypt(tokens.refresh_token),
|
||||
tokenExpiresAt: expiresAt,
|
||||
scope,
|
||||
needsReauth: false,
|
||||
},
|
||||
});
|
||||
|
||||
await auditLog("google_health.connect", {
|
||||
userId: user.id,
|
||||
details: { googleUserId },
|
||||
});
|
||||
|
||||
// Re-completing OAuth clears any prior reauth-required state on the status
|
||||
// ledger too (the `needsReauth` column above is the connection-local flag;
|
||||
// this resets the integration-status streak).
|
||||
await markReconnected(user.id, "google-health");
|
||||
|
||||
// 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(GOOGLE_HEALTH_BACKFILL_QUEUE, {
|
||||
userId: user.id,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
})
|
||||
.catch((err) =>
|
||||
getEvent()?.addWarning(
|
||||
`google-health-backfill enqueue failed: ${err}`,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
const response = NextResponse.redirect(
|
||||
new URL(
|
||||
"/settings/integrations?googleHealth=connected",
|
||||
process.env.NEXT_PUBLIC_APP_URL!,
|
||||
),
|
||||
);
|
||||
response.cookies.delete(GOOGLE_HEALTH_OAUTH_STATE_COOKIE);
|
||||
return response;
|
||||
} catch (err) {
|
||||
getEvent()?.setError(err);
|
||||
return ERR("token");
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,100 @@
|
||||
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 {
|
||||
generatePkcePair,
|
||||
getAuthorizationUrl,
|
||||
} from "@/lib/google-health/client";
|
||||
import { getUserGoogleHealthCredentials } from "@/lib/google-health/credentials";
|
||||
import {
|
||||
GOOGLE_HEALTH_OAUTH_STATE_COOKIE,
|
||||
GOOGLE_HEALTH_OAUTH_STATE_TTL_MS,
|
||||
mintGoogleHealthOAuthStateNonce,
|
||||
} from "@/lib/google-health/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 the Google Health
|
||||
* integration (v1.26.0).
|
||||
*
|
||||
* Mirrors the Fitbit / WHOOP connect route with PKCE added: a fully-random
|
||||
* base64url state nonce backed by a 10-minute `GoogleHealthOAuthState` ledger
|
||||
* row carries the `(nonce → userId)` mapping AND the PKCE `code_verifier`, so
|
||||
* neither the user id nor the verifier ever 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` and reads the verifier off the same row.
|
||||
*
|
||||
* 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: "google_health.connect" } });
|
||||
|
||||
const rl = await checkRateLimit(
|
||||
`google-health:connect:${user.id}`,
|
||||
CONNECT_RATE_LIMIT,
|
||||
CONNECT_WINDOW_MS,
|
||||
);
|
||||
if (!rl.allowed) {
|
||||
annotate({ action: { name: "google_health.connect.rate_limited" } });
|
||||
return NextResponse.redirect(
|
||||
new URL(
|
||||
"/settings/integrations?googleHealth=error&reason=rate_limited",
|
||||
req.url,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
const creds = await getUserGoogleHealthCredentials(user.id);
|
||||
if (!creds) {
|
||||
return apiError(
|
||||
"Please configure your Google OAuth Client ID and Client Secret in Settings first.",
|
||||
400,
|
||||
);
|
||||
}
|
||||
|
||||
const nonce = mintGoogleHealthOAuthStateNonce();
|
||||
const { verifier, challenge } = generatePkcePair();
|
||||
try {
|
||||
await prisma.googleHealthOAuthState.create({
|
||||
data: {
|
||||
nonce,
|
||||
userId: user.id,
|
||||
codeVerifier: verifier,
|
||||
expiresAt: new Date(Date.now() + GOOGLE_HEALTH_OAUTH_STATE_TTL_MS),
|
||||
},
|
||||
});
|
||||
} catch (err) {
|
||||
getEvent()?.setError(err);
|
||||
annotate({ action: { name: "google_health.connect.create_failed" } });
|
||||
return NextResponse.redirect(
|
||||
new URL(
|
||||
"/settings/integrations?googleHealth=error&reason=connect",
|
||||
req.url,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
const url = getAuthorizationUrl(nonce, creds, challenge);
|
||||
|
||||
const response = NextResponse.redirect(url);
|
||||
response.cookies.set(GOOGLE_HEALTH_OAUTH_STATE_COOKIE, nonce, {
|
||||
httpOnly: true,
|
||||
secure: shouldEmitSecureCookie(),
|
||||
sameSite: "lax",
|
||||
maxAge: Math.floor(GOOGLE_HEALTH_OAUTH_STATE_TTL_MS / 1000),
|
||||
path: "/",
|
||||
});
|
||||
|
||||
return response;
|
||||
});
|
||||
@@ -0,0 +1,103 @@
|
||||
import { prisma } from "@/lib/db";
|
||||
import { apiHandler, requireAuth } from "@/lib/api-handler";
|
||||
import { auditLog } from "@/lib/auth/audit";
|
||||
import { apiSuccess, apiError, safeJson } from "@/lib/api-response";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { encrypt } from "@/lib/crypto";
|
||||
import { markDisconnected } from "@/lib/integrations/status";
|
||||
import { NextRequest } from "next/server";
|
||||
import { z } from "zod/v4";
|
||||
|
||||
/**
|
||||
* Per-user 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`.
|
||||
*/
|
||||
const googleHealthCredentialsSchema = z.object({
|
||||
clientId: z.string().min(1).max(200),
|
||||
clientSecret: z.string().min(1).max(200),
|
||||
});
|
||||
|
||||
/**
|
||||
* Check whether the user has Google Health BYO-key credentials stored
|
||||
* (v1.26.0).
|
||||
*/
|
||||
export const GET = apiHandler(async () => {
|
||||
const { user } = await requireAuth();
|
||||
annotate({ action: { name: "google_health.credentials.get" } });
|
||||
|
||||
const dbUser = await prisma.user.findUnique({
|
||||
where: { id: user.id },
|
||||
select: {
|
||||
googleHealthClientIdEncrypted: true,
|
||||
googleHealthClientSecretEncrypted: true,
|
||||
},
|
||||
});
|
||||
|
||||
return apiSuccess({
|
||||
hasCredentials:
|
||||
!!dbUser?.googleHealthClientIdEncrypted &&
|
||||
!!dbUser?.googleHealthClientSecretEncrypted,
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Save Google Health API credentials (encrypted at rest).
|
||||
*/
|
||||
export const PUT = apiHandler(async (request: NextRequest) => {
|
||||
const { user } = await requireAuth();
|
||||
annotate({ action: { name: "google_health.credentials.update" } });
|
||||
|
||||
const { data: body, error: jsonError } = await safeJson(request, {
|
||||
maxBytes: 64 * 1024,
|
||||
});
|
||||
if (jsonError) return jsonError;
|
||||
|
||||
const result = z.safeParse(googleHealthCredentialsSchema, body);
|
||||
if (!result.success) {
|
||||
return apiError("Client ID and Client Secret are required", 422);
|
||||
}
|
||||
|
||||
// Field-by-field assignment — never spread the parsed object, so no future
|
||||
// schema addition can silently mass-assign an unexpected column.
|
||||
await prisma.user.update({
|
||||
where: { id: user.id },
|
||||
data: {
|
||||
googleHealthClientIdEncrypted: encrypt(result.data.clientId),
|
||||
googleHealthClientSecretEncrypted: encrypt(result.data.clientSecret),
|
||||
},
|
||||
});
|
||||
|
||||
return apiSuccess({ updated: true });
|
||||
});
|
||||
|
||||
/**
|
||||
* Delete Google Health credentials and the active connection.
|
||||
*
|
||||
* Audits the teardown and parks the integration ledger at `disconnected` for
|
||||
* parity with the Fitbit / Polar / Oura credential-DELETE so a sensitive op
|
||||
* leaves a uniform audit trail and the status snapshot does not linger at a
|
||||
* stale state.
|
||||
*/
|
||||
export const DELETE = apiHandler(async () => {
|
||||
const { user } = await requireAuth();
|
||||
annotate({ action: { name: "google_health.credentials.delete" } });
|
||||
|
||||
await prisma.googleHealthConnection
|
||||
.delete({ where: { userId: user.id } })
|
||||
.catch(() => {});
|
||||
|
||||
await prisma.user.update({
|
||||
where: { id: user.id },
|
||||
data: {
|
||||
googleHealthClientIdEncrypted: null,
|
||||
googleHealthClientSecretEncrypted: null,
|
||||
},
|
||||
});
|
||||
await auditLog("google_health.credentials.delete", { userId: user.id });
|
||||
await markDisconnected(user.id, "google-health");
|
||||
|
||||
return apiSuccess({ deleted: true });
|
||||
});
|
||||
@@ -0,0 +1,42 @@
|
||||
import { prisma } from "@/lib/db";
|
||||
import { apiHandler, requireAuth } from "@/lib/api-handler";
|
||||
import { auditLog } from "@/lib/auth/audit";
|
||||
import { apiSuccess, apiError } from "@/lib/api-response";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { markDisconnected } from "@/lib/integrations/status";
|
||||
|
||||
/**
|
||||
* Disconnect the Google Health integration for the current user (v1.26.0).
|
||||
*
|
||||
* Clears the stored OAuth tokens by deleting the `GoogleHealthConnection` row
|
||||
* and parks the integration status at `disconnected`. There is no per-user
|
||||
* subscription to tear down (Google Health is poll-only at launch — no Pub/Sub
|
||||
* push), so no upstream unsubscribe call is needed; deleting the encrypted
|
||||
* token pair is the effective revocation. The BYO-key credentials on `User` are
|
||||
* left intact so a reconnect doesn't force the user to re-paste them; use the
|
||||
* credentials DELETE endpoint to remove those too.
|
||||
*/
|
||||
export const POST = apiHandler(async () => {
|
||||
const { user } = await requireAuth();
|
||||
annotate({ action: { name: "google_health.disconnect" } });
|
||||
|
||||
const connection = await prisma.googleHealthConnection.findUnique({
|
||||
where: { userId: user.id },
|
||||
});
|
||||
|
||||
if (!connection) {
|
||||
return apiError("No Google Health connection", 404);
|
||||
}
|
||||
|
||||
await prisma.googleHealthConnection.delete({
|
||||
where: { userId: user.id },
|
||||
});
|
||||
|
||||
await auditLog("google_health.disconnect", {
|
||||
userId: user.id,
|
||||
});
|
||||
|
||||
await markDisconnected(user.id, "google-health");
|
||||
|
||||
return apiSuccess({ disconnected: true });
|
||||
});
|
||||
@@ -0,0 +1,62 @@
|
||||
import { prisma } from "@/lib/db";
|
||||
import { apiHandler, requireAuth } from "@/lib/api-handler";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { apiSuccess } from "@/lib/api-response";
|
||||
|
||||
/**
|
||||
* Get Google Health connection status for the current user (v1.26.0).
|
||||
*
|
||||
* Mirrors the Fitbit status route: `configured` checks the per-user BYO-key
|
||||
* credentials; the connection row reports last-sync + token expiry + backfill
|
||||
* progress. `tokenExpired` is reported straight off the row; the sync path
|
||||
* refreshes lazily via `getValidToken`. `needsReauth` surfaces the soft
|
||||
* disconnect (a refresh that failed with `invalid_grant` — the Testing-mode
|
||||
* 7-day refresh expiry) so the card can render a reconnect CTA.
|
||||
*/
|
||||
export const GET = apiHandler(async () => {
|
||||
const { user } = await requireAuth();
|
||||
annotate({ action: { name: "google_health.status" } });
|
||||
|
||||
const dbUser = await prisma.user.findUnique({
|
||||
where: { id: user.id },
|
||||
select: {
|
||||
googleHealthClientIdEncrypted: true,
|
||||
googleHealthClientSecretEncrypted: true,
|
||||
},
|
||||
});
|
||||
|
||||
const configured =
|
||||
!!dbUser?.googleHealthClientIdEncrypted &&
|
||||
!!dbUser?.googleHealthClientSecretEncrypted;
|
||||
|
||||
const connection = await prisma.googleHealthConnection.findUnique({
|
||||
where: { userId: user.id },
|
||||
select: {
|
||||
googleUserId: true,
|
||||
lastSyncedAt: true,
|
||||
tokenExpiresAt: true,
|
||||
backfillCompletedAt: true,
|
||||
needsReauth: true,
|
||||
createdAt: true,
|
||||
scope: true,
|
||||
},
|
||||
});
|
||||
|
||||
if (!connection) {
|
||||
return apiSuccess({ connected: false, configured });
|
||||
}
|
||||
|
||||
const tokenExpired = connection.tokenExpiresAt.getTime() <= Date.now();
|
||||
|
||||
return apiSuccess({
|
||||
connected: true,
|
||||
configured,
|
||||
lastSyncedAt: connection.lastSyncedAt,
|
||||
connectedAt: connection.createdAt,
|
||||
tokenExpired,
|
||||
tokenExpiresAt: connection.tokenExpiresAt,
|
||||
backfillCompleted: !!connection.backfillCompletedAt,
|
||||
needsReauth: connection.needsReauth,
|
||||
scope: connection.scope,
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,60 @@
|
||||
import { apiHandler, requireAuth } from "@/lib/api-handler";
|
||||
import { apiSuccess, apiError } from "@/lib/api-response";
|
||||
import { checkRateLimit } from "@/lib/rate-limit";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { syncUserGoogleHealth } from "@/lib/google-health/sync";
|
||||
import { NextRequest } from "next/server";
|
||||
|
||||
/**
|
||||
* Manually trigger a Google Health sync for the current user (v1.26.0).
|
||||
*
|
||||
* Mirrors the Fitbit manual-sync route: incremental by default, full history
|
||||
* when `{ fullSync: true }` is posted.
|
||||
*
|
||||
* Rate-limited: a baseline 5/60s bucket gates the route, and the expensive
|
||||
* `fullSync` path (which drives paginated Google Health walkers across every
|
||||
* data type, each capped at 1000 pages) carries a tighter 1/hour bucket of its
|
||||
* own so a re-trigger loop cannot pin Prisma or churn the Google Health quota.
|
||||
*/
|
||||
export const POST = apiHandler(async (request: NextRequest) => {
|
||||
const { user } = await requireAuth();
|
||||
annotate({ action: { name: "google_health.sync" } });
|
||||
|
||||
const rl = await checkRateLimit(`google-health-sync:${user.id}`, 5, 60_000);
|
||||
if (!rl.allowed) {
|
||||
return apiError("Too many sync requests", 429, {
|
||||
errorCode: "rate_limited_self",
|
||||
});
|
||||
}
|
||||
|
||||
let fullSync = false;
|
||||
try {
|
||||
const raw = await request.text();
|
||||
// Flag-only payload — cap the parse cost (mirrors safeJson maxBytes).
|
||||
if (raw.length > 64 * 1024) {
|
||||
return apiError(`Request body exceeds ${64 * 1024} bytes`, 413);
|
||||
}
|
||||
const body = JSON.parse(raw);
|
||||
fullSync = body?.fullSync === true;
|
||||
} catch {
|
||||
// no body provided -> default incremental sync
|
||||
}
|
||||
|
||||
// The full-history walk is the expensive path; cap it well below the
|
||||
// incremental bucket so a re-trigger loop cannot churn the Google quota.
|
||||
if (fullSync) {
|
||||
const fullRl = await checkRateLimit(
|
||||
`google-health-sync-full:${user.id}`,
|
||||
1,
|
||||
60 * 60_000,
|
||||
);
|
||||
if (!fullRl.allowed) {
|
||||
return apiError("Full sync is limited to once per hour", 429, {
|
||||
errorCode: "rate_limited_self",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const imported = await syncUserGoogleHealth(user.id, { fullSync });
|
||||
return apiSuccess({ imported, fullSync });
|
||||
});
|
||||
@@ -55,6 +55,9 @@ export const ENCRYPTED_COLUMNS: readonly EncryptedColumn[] = [
|
||||
{ model: "User", field: "whoopClientSecretEncrypted", kind: "string" },
|
||||
{ model: "User", field: "fitbitClientIdEncrypted", kind: "string" },
|
||||
{ model: "User", field: "fitbitClientSecretEncrypted", kind: "string" },
|
||||
// v1.27.0 — Google Health (Fitbit / Pixel Watch) BYO OAuth client.
|
||||
{ model: "User", field: "googleHealthClientIdEncrypted", kind: "string" },
|
||||
{ model: "User", field: "googleHealthClientSecretEncrypted", kind: "string" },
|
||||
{ model: "User", field: "nightscoutUrlEncrypted", kind: "string" },
|
||||
{ model: "User", field: "nightscoutTokenEncrypted", kind: "string" },
|
||||
{ model: "User", field: "polarAccessTokenEncrypted", kind: "string" },
|
||||
@@ -82,6 +85,8 @@ export const ENCRYPTED_COLUMNS: readonly EncryptedColumn[] = [
|
||||
{ model: "WhoopConnection", field: "refreshToken", kind: "string" },
|
||||
{ model: "FitbitConnection", field: "accessToken", kind: "string" },
|
||||
{ model: "FitbitConnection", field: "refreshToken", kind: "string" },
|
||||
{ model: "GoogleHealthConnection", field: "accessToken", kind: "string" },
|
||||
{ model: "GoogleHealthConnection", field: "refreshToken", kind: "string" },
|
||||
|
||||
// ───── AppSettings — operator credentials ─────
|
||||
{ model: "AppSettings", field: "adminAiKeyEncrypted", kind: "string" },
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
/**
|
||||
* v1.26.0 — pg-boss queue registration for the Google Health sync layer.
|
||||
*
|
||||
* The reminder-worker imports heavy infrastructure (pg-boss, Prisma adapter,
|
||||
* notification dispatchers, …) so we don't boot the whole worker here. Instead
|
||||
* we read the source as text and assert the queue constants + cron schedules +
|
||||
* handler wiring + allQueues + boot-enqueue are present — the fast guard that
|
||||
* catches the v1.4.37 dead-queue class (a queue declared but never registered
|
||||
* in `allQueues`, which silently never drains).
|
||||
*
|
||||
* Google Health is poll-only at launch (no webhook — Pub/Sub deferred), so a
|
||||
* single hourly `google-health-sync` queue drives the per-user
|
||||
* `syncUserGoogleHealth` driver across every connection, alongside the backfill
|
||||
* + the daily OAuth-state sweep.
|
||||
*/
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
const REGISTRAR_PATH = join(
|
||||
__dirname,
|
||||
"..",
|
||||
"reminder",
|
||||
"register-integration-sync.ts",
|
||||
);
|
||||
const source =
|
||||
readFileSync(REGISTRAR_PATH, "utf8") +
|
||||
readFileSync(
|
||||
join(__dirname, "..", "reminder", "google-health-sync.ts"),
|
||||
"utf8",
|
||||
);
|
||||
|
||||
const GOOGLE_HEALTH_QUEUE_CONSTS = [
|
||||
"GOOGLE_HEALTH_SYNC_QUEUE",
|
||||
"GOOGLE_HEALTH_BACKFILL_QUEUE",
|
||||
"GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE",
|
||||
] as const;
|
||||
|
||||
describe("reminder-worker — Google Health sync queues", () => {
|
||||
it("declares the Google Health sync + oauth-state-cleanup queue constants", () => {
|
||||
expect(source).toMatch(
|
||||
/GOOGLE_HEALTH_SYNC_QUEUE\s*=\s*["']google-health-sync["']/,
|
||||
);
|
||||
expect(source).toMatch(
|
||||
/GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE\s*=\s*["']google-health-oauth-state-cleanup["']/,
|
||||
);
|
||||
});
|
||||
|
||||
it("registers EVERY Google Health queue in allQueues (v1.4.37 dead-queue guard)", () => {
|
||||
const block = /const allQueues = \[([\s\S]*?)\];/.exec(source);
|
||||
expect(block).not.toBeNull();
|
||||
const body = block![1]!;
|
||||
for (const q of GOOGLE_HEALTH_QUEUE_CONSTS) {
|
||||
expect(body).toContain(q);
|
||||
}
|
||||
});
|
||||
|
||||
it("schedules the hourly poll cron + the daily oauth-state sweep", () => {
|
||||
expect(source).toMatch(
|
||||
/\[GOOGLE_HEALTH_SYNC_QUEUE,\s*GOOGLE_HEALTH_SYNC_CRON\]/,
|
||||
);
|
||||
expect(source).toMatch(
|
||||
/\[\s*GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE,\s*GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_CRON,?\s*\]/,
|
||||
);
|
||||
});
|
||||
|
||||
it("registers a boss.work handler for every Google Health queue", () => {
|
||||
expect(source).toMatch(
|
||||
/boss\.work[\s\S]{0,200}GOOGLE_HEALTH_SYNC_QUEUE[\s\S]{0,200}handleGoogleHealthSync/,
|
||||
);
|
||||
expect(source).toMatch(
|
||||
/boss\.work[\s\S]{0,320}GOOGLE_HEALTH_BACKFILL_QUEUE[\s\S]{0,320}runGoogleHealthBackfillForUser/,
|
||||
);
|
||||
expect(source).toMatch(
|
||||
/boss\.work[\s\S]{0,200}GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE[\s\S]{0,200}handleGoogleHealthOAuthStateCleanup/,
|
||||
);
|
||||
});
|
||||
|
||||
it("wires the self-converging Google Health backfill boot discovery", () => {
|
||||
expect(source).toMatch(/enqueueBootTimeGoogleHealthBackfill\(\)/);
|
||||
});
|
||||
|
||||
it("imports the Google Health sync driver + backfill exports", () => {
|
||||
expect(source).toMatch(/from\s*["']@\/lib\/google-health\/sync["']/);
|
||||
expect(source).toMatch(
|
||||
/from\s*["']@\/lib\/jobs\/google-health-backfill["']/,
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,117 @@
|
||||
/**
|
||||
* v1.26.0 — pg-boss queue + boot-time self-converging backfill for newly
|
||||
* connected Google Health accounts. Modelled on the Fitbit / WHOOP backfill
|
||||
* (`src/lib/jobs/fitbit-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 queue name MUST be registered in `allQueues` in
|
||||
* `src/lib/jobs/reminder/register-integration-sync.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 { annotate } from "@/lib/logging/context";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { syncUserGoogleHealth } from "@/lib/google-health/sync";
|
||||
|
||||
export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-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 `FITBIT_BACKFILL_CONCURRENCY`.
|
||||
*/
|
||||
export const GOOGLE_HEALTH_BACKFILL_CONCURRENCY = 1;
|
||||
|
||||
export interface GoogleHealthBackfillPayload {
|
||||
userId: string;
|
||||
enqueuedAt: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-user backfill handler. Runs a full-history sync for one account and
|
||||
* stamps `backfillCompletedAt` so the discovery query drops it. Idempotent: the
|
||||
* per-resource upserts are key-stable, so a re-run (e.g. a reboot mid-walk)
|
||||
* overwrites rather than duplicating. Mirrors `runFitbitBackfillForUser`.
|
||||
*/
|
||||
export async function runGoogleHealthBackfillForUser(
|
||||
userId: string,
|
||||
): Promise<{ imported: number }> {
|
||||
const imported = await syncUserGoogleHealth(userId, { fullSync: true });
|
||||
|
||||
await prisma.googleHealthConnection.update({
|
||||
where: { userId },
|
||||
data: { backfillCompletedAt: new Date() },
|
||||
});
|
||||
|
||||
annotate({
|
||||
action: {
|
||||
name: "google_health.backfill.complete",
|
||||
details: { imported },
|
||||
},
|
||||
});
|
||||
return { imported };
|
||||
}
|
||||
|
||||
/**
|
||||
* Boot-time discovery. Finds every Google Health 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 enqueueBootTimeGoogleHealthBackfill(): 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.googleHealthConnection.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: GoogleHealthBackfillPayload = {
|
||||
userId,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
};
|
||||
const jobId = await boss.send(GOOGLE_HEALTH_BACKFILL_QUEUE, payload, {
|
||||
retryLimit: 3,
|
||||
retryDelay: 60,
|
||||
retryBackoff: true,
|
||||
singletonKey: `google-health-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),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
/**
|
||||
* v1.26.0 — daily cleanup for the `google_health_oauth_states` table.
|
||||
*
|
||||
* Mirrors `fitbit-oauth-state-cleanup`. The OAuth state ledger is single-use —
|
||||
* every happy-path + error-path branch in `google-health/callback` consumes its
|
||||
* row, so an abandoned row only lingers when the user closed the Google approval
|
||||
* tab without bouncing back to the callback URL. The cookie + row TTL is 10
|
||||
* minutes, so once the timestamp blows the row is dead weight.
|
||||
*
|
||||
* Idempotent: a re-run within the same window matches zero rows the second time
|
||||
* because the first pass deleted everything older than `now()`.
|
||||
*/
|
||||
import type { PrismaClient } from "@/generated/prisma/client";
|
||||
|
||||
export async function cleanupExpiredGoogleHealthOAuthStates(
|
||||
prisma: PrismaClient,
|
||||
now: Date = new Date(),
|
||||
): Promise<number> {
|
||||
const { count } = await prisma.googleHealthOAuthState.deleteMany({
|
||||
where: { expiresAt: { lt: now } },
|
||||
});
|
||||
return count;
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
/**
|
||||
* Google Health hourly poll-cohort sync and OAuth-state cleanup handlers.
|
||||
*
|
||||
* Extracted alongside the Fitbit handlers; register-integration-sync.ts owns the
|
||||
* queue names, cron schedules, and boss.work registrations.
|
||||
*/
|
||||
import { type Job } from "pg-boss";
|
||||
import { recordError } from "@/lib/jobs/worker-status";
|
||||
import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
import { runGoogleHealthPollCohort } from "@/lib/google-health/sync";
|
||||
import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
|
||||
import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup";
|
||||
import { getWorkerPrisma } from "./shared";
|
||||
|
||||
/**
|
||||
* v1.26.0 — Google Health poll-sync payload. Poll-only (no webhook — Pub/Sub is
|
||||
* deferred): the single hourly cron tick carries no `userId`, so the handler
|
||||
* iterates every Google Health connection and re-syncs each via
|
||||
* `syncUserGoogleHealth`. One user's parked-at-reauth state never starves the
|
||||
* rest of the cohort.
|
||||
*/
|
||||
export interface GoogleHealthSyncPayload {
|
||||
userId?: string;
|
||||
}
|
||||
|
||||
export async function handleGoogleHealthSync(
|
||||
jobs: Job<GoogleHealthSyncPayload>[],
|
||||
) {
|
||||
await withBackgroundEvent("job.google_health_sync", async (evt) => {
|
||||
const prisma = getWorkerPrisma();
|
||||
try {
|
||||
const targets: Array<{ userId: string }> = [];
|
||||
for (const job of jobs) {
|
||||
if (job.data?.userId) targets.push({ userId: job.data.userId });
|
||||
}
|
||||
if (targets.length === 0) {
|
||||
const connections = await prisma.googleHealthConnection.findMany({
|
||||
select: { userId: true },
|
||||
});
|
||||
targets.push(...connections);
|
||||
}
|
||||
if (targets.length === 0) return;
|
||||
|
||||
// Fan the cohort out with bounded concurrency + per-user error isolation:
|
||||
// one slow Google response can't stall the whole cohort, and a single
|
||||
// user's failure is warned without aborting the pass.
|
||||
const { usersSynced, measurementsImported } =
|
||||
await runGoogleHealthPollCohort(
|
||||
targets.map((t) => t.userId),
|
||||
{
|
||||
onUserError: (userId, err) =>
|
||||
evt.addWarning(
|
||||
`job.google_health_sync failed for user ${userId}: ${err}`,
|
||||
),
|
||||
// A fresh reading landed; resolve the user's Vorsorge reminders
|
||||
// eventfully. Fire-and-forget; the cron is the net.
|
||||
onUserSynced: (userId, imported) => {
|
||||
if (imported > 0) {
|
||||
void enqueueReminderSatisfy(userId).catch(() => {});
|
||||
}
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
evt.setBackground({
|
||||
task_name: "job.google_health_sync",
|
||||
result: {
|
||||
users_synced: usersSynced,
|
||||
total: targets.length,
|
||||
measurements_imported: measurementsImported,
|
||||
},
|
||||
});
|
||||
} catch (err) {
|
||||
evt.setError(err);
|
||||
recordError();
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export interface GoogleHealthOAuthStateCleanupPayload {
|
||||
triggeredAt?: string;
|
||||
}
|
||||
|
||||
export async function handleGoogleHealthOAuthStateCleanup(
|
||||
jobs: Job<GoogleHealthOAuthStateCleanupPayload>[],
|
||||
) {
|
||||
void jobs;
|
||||
await withBackgroundEvent(
|
||||
"job.google_health_oauth_state_cleanup",
|
||||
async (evt) => {
|
||||
const p = getWorkerPrisma();
|
||||
try {
|
||||
const deleted = await cleanupExpiredGoogleHealthOAuthStates(p);
|
||||
evt.addMeta("google_health_oauth_state_cleanup_deleted", deleted);
|
||||
} catch (err) {
|
||||
evt.addWarning(`google-health-oauth-state-cleanup failed: ${err}`);
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -28,6 +28,13 @@ import {
|
||||
enqueueBootTimeFitbitBackfill,
|
||||
type FitbitBackfillPayload,
|
||||
} from "@/lib/jobs/fitbit-backfill";
|
||||
import {
|
||||
GOOGLE_HEALTH_BACKFILL_QUEUE,
|
||||
GOOGLE_HEALTH_BACKFILL_CONCURRENCY,
|
||||
runGoogleHealthBackfillForUser,
|
||||
enqueueBootTimeGoogleHealthBackfill,
|
||||
type GoogleHealthBackfillPayload,
|
||||
} from "@/lib/jobs/google-health-backfill";
|
||||
import {
|
||||
SLEEP_TIMELINE_BACKFILL_QUEUE,
|
||||
SLEEP_TIMELINE_BACKFILL_CONCURRENCY,
|
||||
@@ -75,6 +82,12 @@ import {
|
||||
FitbitOAuthStateCleanupPayload,
|
||||
handleFitbitOAuthStateCleanup,
|
||||
} from "./fitbit-sync";
|
||||
import {
|
||||
GoogleHealthSyncPayload,
|
||||
handleGoogleHealthSync,
|
||||
GoogleHealthOAuthStateCleanupPayload,
|
||||
handleGoogleHealthOAuthStateCleanup,
|
||||
} from "./google-health-sync";
|
||||
|
||||
const WITHINGS_SYNC_QUEUE = "withings-fallback-sync";
|
||||
|
||||
@@ -150,6 +163,23 @@ const FITBIT_SYNC_CRON = "8 * * * *"; // every hour at :08
|
||||
const FITBIT_OAUTH_STATE_CLEANUP_QUEUE = "fitbit-oauth-state-cleanup";
|
||||
|
||||
const FITBIT_OAUTH_STATE_CLEANUP_CRON = "24 3 * * *";
|
||||
// v1.26.0 — Google Health / Fitbit-Pixel-Wear-OS poll-only sync. There is no
|
||||
// Google Health webhook at launch (Pub/Sub deferred), so a single hourly cron
|
||||
// drives the per-user `syncUserGoogleHealth` driver across every connection.
|
||||
// Minute staggered off the WHOOP (:05/:20/:35/:50), Fitbit (:08), Nightscout
|
||||
// (:11), Polar (:13), and Oura (:15) slots so the hourly ticks don't pile up on
|
||||
// one boss poll.
|
||||
|
||||
const GOOGLE_HEALTH_SYNC_QUEUE = "google-health-sync";
|
||||
|
||||
const GOOGLE_HEALTH_SYNC_CRON = "18 * * * *"; // every hour at :18
|
||||
// v1.26.0 — daily sweep for the Google Health OAuth state ledger. Slots at
|
||||
// 03:26, next to the Fitbit sweep (03:24), inside the maintenance window.
|
||||
|
||||
const GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE =
|
||||
"google-health-oauth-state-cleanup";
|
||||
|
||||
const GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_CRON = "26 3 * * *";
|
||||
// v1.17.0 — Nightscout CGM poll sync. Poll-only (no webhook): one hourly tick
|
||||
// pulls the recent SGV window per configured instance. :11 staggers off the
|
||||
// WHOOP (:05), Fitbit (:08), and Withings sync ticks so the hourly polls don't
|
||||
@@ -195,6 +225,11 @@ const allQueues = [
|
||||
FITBIT_SYNC_QUEUE,
|
||||
FITBIT_BACKFILL_QUEUE,
|
||||
FITBIT_OAUTH_STATE_CLEANUP_QUEUE,
|
||||
// v1.26.0 — Google Health poll-only sync (no webhook at launch),
|
||||
// self-converging boot backfill, and the daily OAuth-state ledger sweep.
|
||||
GOOGLE_HEALTH_SYNC_QUEUE,
|
||||
GOOGLE_HEALTH_BACKFILL_QUEUE,
|
||||
GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE,
|
||||
// v1.17.1 — one-shot sleep-timeline backfill for WHOOP + Withings.
|
||||
// Discovery enqueues one job per connection whose sleep rows predate the
|
||||
// stamp/shape fix; the pass deletes the affected SLEEP_DURATION rows and
|
||||
@@ -230,6 +265,14 @@ const schedules: ScheduleEntry[] = [
|
||||
// daily 03:24 Europe/Berlin prune for expired Fitbit OAuth states.
|
||||
[FITBIT_SYNC_QUEUE, FITBIT_SYNC_CRON],
|
||||
[FITBIT_OAUTH_STATE_CLEANUP_QUEUE, FITBIT_OAUTH_STATE_CLEANUP_CRON],
|
||||
// v1.26.0 — hourly Google Health poll (:18, staggered off WHOOP/Fitbit/
|
||||
// Nightscout/Polar/Oura) + the daily 03:26 Europe/Berlin prune for expired
|
||||
// Google Health OAuth states.
|
||||
[GOOGLE_HEALTH_SYNC_QUEUE, GOOGLE_HEALTH_SYNC_CRON],
|
||||
[
|
||||
GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE,
|
||||
GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_CRON,
|
||||
],
|
||||
// v1.17.0 — hourly Nightscout CGM poll (:11, staggered off the other sync
|
||||
// ticks).
|
||||
[NIGHTSCOUT_SYNC_QUEUE, NIGHTSCOUT_SYNC_CRON],
|
||||
@@ -341,6 +384,36 @@ export async function registerIntegrationSyncQueues(
|
||||
{ localConcurrency: 1 },
|
||||
handleFitbitOAuthStateCleanup,
|
||||
);
|
||||
// v1.26.0 — Google Health poll-sync (cron full-iteration; no webhook). Serial
|
||||
// concurrency so a backfill-heavy tick never crowds the request pool.
|
||||
await boss.work<GoogleHealthSyncPayload>(
|
||||
GOOGLE_HEALTH_SYNC_QUEUE,
|
||||
{ localConcurrency: 1 },
|
||||
handleGoogleHealthSync,
|
||||
);
|
||||
// v1.26.0 — self-converging Google Health backfill. The boot enqueue below
|
||||
// sends one full-history sync per un-backfilled connection; this handler runs
|
||||
// it and stamps `backfillCompletedAt` so the discovery query drops the
|
||||
// account.
|
||||
await boss.work<GoogleHealthBackfillPayload>(
|
||||
GOOGLE_HEALTH_BACKFILL_QUEUE,
|
||||
{ localConcurrency: GOOGLE_HEALTH_BACKFILL_CONCURRENCY },
|
||||
async (jobs) => {
|
||||
for (const job of jobs) {
|
||||
const { userId } = job.data;
|
||||
const { imported } = await runGoogleHealthBackfillForUser(userId);
|
||||
workerLog(
|
||||
"info",
|
||||
`[google-health-backfill] user=${userId} imported=${imported}`,
|
||||
);
|
||||
}
|
||||
},
|
||||
);
|
||||
await boss.work<GoogleHealthOAuthStateCleanupPayload>(
|
||||
GOOGLE_HEALTH_OAUTH_STATE_CLEANUP_QUEUE,
|
||||
{ localConcurrency: 1 },
|
||||
handleGoogleHealthOAuthStateCleanup,
|
||||
);
|
||||
// v1.17.1 — one-shot sleep-timeline backfill. The boot enqueue below sends
|
||||
// one job per (user, provider) whose sleep rows predate the stamp/shape fix;
|
||||
// this handler deletes the affected rows, re-syncs, and stamps the marker so
|
||||
@@ -458,6 +531,30 @@ export async function enqueueIntegrationSyncBootDiscovery(): Promise<void> {
|
||||
);
|
||||
}
|
||||
|
||||
// v1.26.0 — Google Health backfill. Finds every Google Health connection not
|
||||
// yet backfilled and enqueues one full-history sync per account.
|
||||
try {
|
||||
const { enqueued, skipped, error } =
|
||||
await enqueueBootTimeGoogleHealthBackfill();
|
||||
if (error) {
|
||||
workerLog(
|
||||
"error",
|
||||
`[google-health-backfill] boot discovery failed: ${error}`,
|
||||
);
|
||||
} else {
|
||||
workerLog(
|
||||
"info",
|
||||
`[google-health-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
workerLog(
|
||||
"error",
|
||||
"[google-health-backfill] boot discovery threw an unexpected error",
|
||||
err,
|
||||
);
|
||||
}
|
||||
|
||||
// v1.17.1 — one-shot sleep-timeline backfill. Finds every WHOOP + Withings
|
||||
// connection whose sleep rows predate the stamp/shape fix and enqueues one
|
||||
// job per (user, provider). A completed pass stamps `sleepTimelineBackfillAt`.
|
||||
|
||||
@@ -204,6 +204,12 @@ export const measurementSourceEnum = z.enum([
|
||||
// of this enum so the read/response shapes (and the iOS decoder) can decode
|
||||
// the rows it surfaces.
|
||||
"MCP",
|
||||
// v1.26.0 — Fitbit/Pixel + Wear OS via the Google Health API. Server-owned
|
||||
// (no client write path), exactly like WHOOP/FITBIT. Part of this enum so the
|
||||
// read/response shapes (and the iOS decoder) can decode the rows it surfaces;
|
||||
// deliberately absent from `WRITABLE_MEASUREMENT_SOURCES` + the batch
|
||||
// allowlist so a client can never forge a GOOGLE_HEALTH-attributed row.
|
||||
"GOOGLE_HEALTH",
|
||||
]);
|
||||
|
||||
/**
|
||||
|
||||
@@ -234,11 +234,24 @@ export const DEFAULT_SOURCE_PRIORITY: Required<MetricPriority> = {
|
||||
// they rank below the phone-aggregated Apple Health (which sees all-day
|
||||
// steps) but as legitimate sources above MANUAL, so an Oura/Polar-only day
|
||||
// is not dropped when another source coexists.
|
||||
steps: ["APPLE_HEALTH", "WITHINGS", "FITBIT", "OURA", "POLAR", "MANUAL"],
|
||||
// v1.26.0 — Google Health surfaces Fitbit/Pixel/Wear OS device data (the
|
||||
// successor API to the classic Fitbit Web API); it rides in the same
|
||||
// wrist-wearable class as FITBIT and ranks immediately below it in every
|
||||
// device ladder, above OURA/POLAR and MANUAL.
|
||||
steps: [
|
||||
"APPLE_HEALTH",
|
||||
"WITHINGS",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"MANUAL",
|
||||
],
|
||||
activeEnergy: [
|
||||
"APPLE_HEALTH",
|
||||
"WITHINGS",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"MANUAL",
|
||||
@@ -247,11 +260,18 @@ export const DEFAULT_SOURCE_PRIORITY: Required<MetricPriority> = {
|
||||
"APPLE_HEALTH",
|
||||
"WITHINGS",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"MANUAL",
|
||||
],
|
||||
flightsClimbed: ["APPLE_HEALTH", "WITHINGS", "FITBIT", "MANUAL"],
|
||||
flightsClimbed: [
|
||||
"APPLE_HEALTH",
|
||||
"WITHINGS",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"MANUAL",
|
||||
],
|
||||
// v1.11.0 — WHOOP leads the recovery-input ladders (sleep / HRV / RHR):
|
||||
// a worn-all-night strap has higher-resolution overnight sampling than
|
||||
// the iPhone-relayed HealthKit summary or the Withings nightly summary.
|
||||
@@ -260,26 +280,64 @@ export const DEFAULT_SOURCE_PRIORITY: Required<MetricPriority> = {
|
||||
// v1.17.0 — Polar + Oura are worn wearables in the same overnight-sampling
|
||||
// class as WHOOP/Fitbit; they rank below the established straps but above
|
||||
// the iPhone-relayed HealthKit summary and the Withings nightly summary.
|
||||
sleep: ["WHOOP", "FITBIT", "OURA", "POLAR", "APPLE_HEALTH", "WITHINGS"],
|
||||
hrv: ["WHOOP", "FITBIT", "OURA", "POLAR", "APPLE_HEALTH", "WITHINGS"],
|
||||
sleep: [
|
||||
"WHOOP",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"APPLE_HEALTH",
|
||||
"WITHINGS",
|
||||
],
|
||||
hrv: [
|
||||
"WHOOP",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"APPLE_HEALTH",
|
||||
"WITHINGS",
|
||||
],
|
||||
restingHeartRate: [
|
||||
"WHOOP",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"APPLE_HEALTH",
|
||||
"WITHINGS",
|
||||
],
|
||||
// A real scale beats a strap's body-measurement estimate for weight.
|
||||
weight: ["WITHINGS", "APPLE_HEALTH", "MANUAL", "WHOOP", "FITBIT"],
|
||||
weight: [
|
||||
"WITHINGS",
|
||||
"APPLE_HEALTH",
|
||||
"MANUAL",
|
||||
"WHOOP",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
],
|
||||
bloodPressure: ["WITHINGS", "APPLE_HEALTH", "MANUAL"],
|
||||
// v1.17.0 — continuous optical-HR wearables (Fitbit / Polar / Oura) rank
|
||||
// above a hand-typed MANUAL pulse; the device-source order matches the
|
||||
// rhr / hrv ladders (FITBIT > OURA > POLAR). Withings BPM cuff + the
|
||||
// HealthKit relay stay the primary point-measurement sources.
|
||||
pulse: ["WITHINGS", "APPLE_HEALTH", "FITBIT", "OURA", "POLAR", "MANUAL"],
|
||||
bodyFat: ["WITHINGS", "APPLE_HEALTH", "MANUAL", "FITBIT"],
|
||||
bodyTemperature: ["WITHINGS", "APPLE_HEALTH", "MANUAL", "FITBIT"],
|
||||
pulse: [
|
||||
"WITHINGS",
|
||||
"APPLE_HEALTH",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"MANUAL",
|
||||
],
|
||||
bodyFat: ["WITHINGS", "APPLE_HEALTH", "MANUAL", "FITBIT", "GOOGLE_HEALTH"],
|
||||
bodyTemperature: [
|
||||
"WITHINGS",
|
||||
"APPLE_HEALTH",
|
||||
"MANUAL",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
],
|
||||
// Withings ScanWatch pulse-ox is the primary SpO2 sensor; WHOOP second,
|
||||
// Fitbit third, Oura's ring fourth. v1.17.1 — Polar Elixir pulse-ox slots
|
||||
// below the established straps and above the iPhone-relay / manual sources.
|
||||
@@ -287,19 +345,35 @@ export const DEFAULT_SOURCE_PRIORITY: Required<MetricPriority> = {
|
||||
"WITHINGS",
|
||||
"WHOOP",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"APPLE_HEALTH",
|
||||
"MANUAL",
|
||||
],
|
||||
vo2Max: ["WITHINGS", "APPLE_HEALTH", "FITBIT", "OURA", "MANUAL"],
|
||||
vo2Max: [
|
||||
"WITHINGS",
|
||||
"APPLE_HEALTH",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"MANUAL",
|
||||
],
|
||||
// v1.11.0 — new WHOOP-overlapping keys. ScanWatch dermal reading is the
|
||||
// primary skin-temperature sensor; WHOOP's strap is second, Fitbit third,
|
||||
// Oura's ring fourth.
|
||||
skinTemperature: ["WITHINGS", "WHOOP", "FITBIT", "OURA", "APPLE_HEALTH"],
|
||||
skinTemperature: [
|
||||
"WITHINGS",
|
||||
"WHOOP",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"APPLE_HEALTH",
|
||||
],
|
||||
respiratoryRate: [
|
||||
"WHOOP",
|
||||
"FITBIT",
|
||||
"GOOGLE_HEALTH",
|
||||
"OURA",
|
||||
"POLAR",
|
||||
"APPLE_HEALTH",
|
||||
|
||||
Reference in new issue
Block a user