fix(integrations): refresh-token failure marks integration as needing re-auth

v1.4.15 Phase B2, criterion 2. The Phase B2 status helper (commit
4db72a8) defined a markReauthRequired / recordSyncFailure API but
nothing called it. A Withings refresh token that had revoked got the
same "transient error" treatment as a 503 from the upstream API,
scheduled syncs kept hammering the OAuth endpoint, and the Settings
UI showed "Connected" until the user manually triggered a sync and
read the inline error.

This commit wires every sync / OAuth path:

  - getValidToken() (Withings): refresh failures classified by
    Withings status code. 100/101/102 (auth failed / invalid token /
    user gone) and 200..299 (OAuth grant errors) -> error_reauth.
    Anything else (5xx, network) -> error_transient. Same logic
    inlined in /api/withings/status so the proactive refresh path
    parks the row too.
  - syncUserMeasurements() and syncMoodLogEntries() short-circuit
    immediately when isReauthRequired() returns true. Scheduled
    pg-boss jobs no longer burn API quota on a parked connection.
  - moodLog 401/403 from the upstream -> error_reauth. 5xx / 3xx
    redirect / network / invalid-JSON / bad-format -> error_transient.
    SSRF refusal also writes a status row so it surfaces in the UI.
  - /api/withings/callback calls markReconnected() after writing the
    new tokens, the next sync runs.
  - /api/withings/disconnect and DELETE /api/settings/moodlog call
    markDisconnected() to clear any prior error state. moodLog
    PUT /api/settings/moodlog (re-save credentials) markReconnected()s
    for the same reason.

The helper writes AuditLog action=integrations.sync.failed on every
failure (criterion 4) and dispatches a SYSTEM_ALERT notification to
admins after consecutiveFailures >= 3 with a 24h re-alert window
(criterion 3). Both paths are covered by the unit + integration test
suites that landed in 4db72a8 — this commit makes them live.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/app/api/settings/moodlog/route.ts

@@ -7,6 +7,7 @@ import { encryptMoodLogSecret, readMoodLogSecret } from "@/lib/moodlog-secret";
 import { moodLogCredentialsSchema } from "@/lib/validations/moodlog";
 import { apiHandler, requireAuth } from "@/lib/api-handler";
 import { annotate } from "@/lib/logging/context";
+import { markDisconnected, markReconnected } from "@/lib/integrations/status";
 
 export const dynamic = "force-dynamic";
 
@@ -53,6 +54,11 @@ export const PUT = apiHandler(async (request: NextRequest) => {
 
   annotate({ action: { name: "settings.moodlog.update" } });
 
+  // Re-saving credentials clears any prior `error_reauth` state so the
+  // next scheduled sync can run again. We don't write a fresh
+  // success-time — that happens on the first real sync.
+  await markReconnected(user.id, "moodlog");
+
   return apiSuccess({
     configured: true,
     enabled: true,
@@ -82,5 +88,7 @@ export const DELETE = apiHandler(async () => {
 
   annotate({ action: { name: "settings.moodlog.disconnect" } });
 
+  await markDisconnected(user.id, "moodlog");
+
   return apiSuccess({ disconnected: true });
 });

src/app/api/withings/callback/route.ts

@@ -6,6 +6,7 @@ import { encrypt } from "@/lib/crypto";
 import { exchangeCode } from "@/lib/withings/client";
 import { getUserWithingsCredentials } from "@/lib/withings/credentials";
 import { setupWebhook } from "@/lib/withings/sync";
+import { markReconnected } from "@/lib/integrations/status";
 import { NextRequest, NextResponse } from "next/server";
 import { timingSafeEqual } from "node:crypto";
 
@@ -99,6 +100,10 @@ export const GET = apiHandler(async (request: NextRequest) => {
       details: { withingsUserId: tokens.userid },
     });
 
+    // Re-completing OAuth clears any prior `error_reauth` state. We
+    // don't write a fresh success-time — that's the next sync's job.
+    await markReconnected(user.id, "withings");
+
     const response = NextResponse.redirect(
       new URL(
         "/settings/integrations?withings=connected",

src/app/api/withings/disconnect/route.ts

@@ -6,6 +6,7 @@ import { annotate } from "@/lib/logging/context";
 import { decrypt } from "@/lib/crypto";
 import { unsubscribeWebhook } from "@/lib/withings/client";
 import { getWithingsWebhookCallbackUrl } from "@/lib/withings/sync";
+import { markDisconnected } from "@/lib/integrations/status";
 
 /**
  * Disconnect Withings integration for the current user.
@@ -39,5 +40,7 @@ export const POST = apiHandler(async () => {
     userId: user.id,
   });
 
+  await markDisconnected(user.id, "withings");
+
   return apiSuccess({ disconnected: true });
 });

src/app/api/withings/status/route.ts

@@ -5,6 +5,7 @@ import { apiSuccess } from "@/lib/api-response";
 import { decrypt, encrypt } from "@/lib/crypto";
 import { getUserWithingsCredentials } from "@/lib/withings/credentials";
 import { refreshAccessToken } from "@/lib/withings/client";
+import { markReauthRequired } from "@/lib/integrations/status";
 
 /**
  * Get Withings connection status for the current user.
@@ -75,6 +76,23 @@ export const GET = apiHandler(async () => {
         tokenRefreshFailed = true;
       }
       getEvent()?.addWarning("Token refresh failed");
+      // A permanently revoked grant should park the integration until
+      // the user reconnects. We mirror the same status-code parsing
+      // the sync path does.
+      const message = error instanceof Error ? error.message : String(error);
+      const statusMatch = /Withings\s+\w+\s+error:\s*(\d+)/.exec(message);
+      const statusCode = statusMatch
+        ? Number.parseInt(statusMatch[1], 10)
+        : NaN;
+      const isReauth =
+        Number.isFinite(statusCode) &&
+        (statusCode === 100 ||
+          statusCode === 101 ||
+          statusCode === 102 ||
+          (statusCode >= 200 && statusCode <= 299));
+      if (isReauth) {
+        await markReauthRequired(user.id, "withings", message).catch(() => {});
+      }
     }
   }
 

src/lib/moodlog/sync.ts

@@ -2,6 +2,11 @@ import { prisma } from "@/lib/db";
 import { decrypt } from "@/lib/crypto";
 import { getEvent } from "@/lib/logging/context";
 import { isPublicUrl } from "@/lib/validations/notifications";
+import {
+  isReauthRequired,
+  recordSyncFailure,
+  recordSyncSuccess,
+} from "@/lib/integrations/status";
 
 /**
  * Sync mood entries from a user's moodLog instance.
@@ -30,6 +35,16 @@ export async function syncMoodLogEntries(
     return 0;
   }
 
+  // Park: if we already know the apiKey is rejected, skip the network
+  // round-trip until the user re-saves credentials. The settings PUT
+  // route clears this state (see /api/settings/moodlog).
+  if (await isReauthRequired(userId, "moodlog")) {
+    getEvent()?.addWarning(
+      `moodlog sync skipped for ${userId}: parked at error_reauth`,
+    );
+    return 0;
+  }
+
   const baseUrl = decrypt(user.moodLogUrlEncrypted);
   const apiKey = decrypt(user.moodLogApiKeyEncrypted);
 
@@ -42,6 +57,13 @@ export async function syncMoodLogEntries(
     getEvent()?.addWarning(
       `moodLog sync refused for user ${userId}: stored URL points at non-public host`,
     );
+    await recordSyncFailure({
+      userId,
+      integration: "moodlog",
+      kind: "reauth_required",
+      message: "Stored moodLog URL points at a non-public host",
+      errorCode: "ssrf_refused",
+    });
     return 0;
   }
 
@@ -79,20 +101,44 @@ export async function syncMoodLogEntries(
       // the apiKey to the internal hop.
       redirect: "manual",
     });
-  } finally {
+  } catch (err) {
     clearTimeout(timeout);
     getEvent()?.addExternalCall({
       service: "moodlog",
       method: "syncMoodLogEntries",
       duration_ms: Math.round(performance.now() - fetchStart),
     });
+    const message = err instanceof Error ? err.message : String(err);
+    await recordSyncFailure({
+      userId,
+      integration: "moodlog",
+      kind: "transient",
+      message,
+      errorCode: "fetch_failed",
+    });
+    return 0;
+  } finally {
+    clearTimeout(timeout);
   }
+  getEvent()?.addExternalCall({
+    service: "moodlog",
+    method: "syncMoodLogEntries",
+    duration_ms: Math.round(performance.now() - fetchStart),
+    status: response.status,
+  });
 
   // Treat any 3xx as failure (manual redirect mode surfaces them).
   if (response.status >= 300 && response.status < 400) {
     getEvent()?.addWarning(
       `moodLog sync refused redirect for user ${userId}: HTTP ${response.status}`,
     );
+    await recordSyncFailure({
+      userId,
+      integration: "moodlog",
+      kind: "transient",
+      message: `moodLog sync refused redirect (HTTP ${response.status})`,
+      errorCode: `http_${response.status}`,
+    });
     return 0;
   }
 
@@ -100,23 +146,66 @@ export async function syncMoodLogEntries(
     getEvent()?.addWarning(
       `Sync failed for user ${userId}: HTTP ${response.status}`,
     );
+    // 401/403 — apiKey rejected. Park the integration so we don't
+    // hammer the upstream until the user re-saves credentials.
+    const isAuthFailure = response.status === 401 || response.status === 403;
+    await recordSyncFailure({
+      userId,
+      integration: "moodlog",
+      kind: isAuthFailure ? "reauth_required" : "transient",
+      message: `moodLog sync HTTP ${response.status}`,
+      errorCode: `http_${response.status}`,
+    });
     return 0;
   }
 
-  const data = await response.json();
+  let data: unknown;
+  try {
+    data = await response.json();
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    await recordSyncFailure({
+      userId,
+      integration: "moodlog",
+      kind: "transient",
+      message: `moodLog sync invalid JSON: ${message}`,
+      errorCode: "invalid_json",
+    });
+    return 0;
+  }
 
   // Validate response has entries array
-  if (!data?.entries || !Array.isArray(data.entries)) {
+  if (
+    !data ||
+    typeof data !== "object" ||
+    !("entries" in data) ||
+    !Array.isArray((data as { entries?: unknown }).entries)
+  ) {
     getEvent()?.addWarning(`Invalid response format for user ${userId}`);
+    await recordSyncFailure({
+      userId,
+      integration: "moodlog",
+      kind: "transient",
+      message: "moodLog sync invalid response format (missing 'entries' array)",
+      errorCode: "invalid_format",
+    });
     return 0;
   }
+  const entries = (data as { entries: unknown[] }).entries;
 
   // 4. Upsert entries (note field is intentionally NOT imported)
   let imported = 0;
-  for (const entry of data.entries) {
+  for (const e of entries) {
+    const entry = e as {
+      time: string;
+      date: string;
+      mood: string;
+      score: number;
+      tags?: string[];
+    };
     try {
       const moodLoggedAt = new Date(entry.time);
-      const date = entry.date as string;
+      const date = entry.date;
 
       await prisma.moodEntry.upsert({
         where: {
@@ -148,11 +237,12 @@ export async function syncMoodLogEntries(
     }
   }
 
-  // 5. Update lastSyncedAt
+  // 5. Update lastSyncedAt + clear sync-status streak
   await prisma.user.update({
     where: { id: userId },
     data: { moodLogLastSyncedAt: now },
   });
+  await recordSyncSuccess(userId, "moodlog");
 
   return imported;
 }

src/lib/withings/sync.ts

@@ -13,6 +13,11 @@ import {
 } from "./client";
 import { getUserWithingsCredentials } from "./credentials";
 import { getEvent } from "@/lib/logging/context";
+import {
+  isReauthRequired,
+  recordSyncFailure,
+  recordSyncSuccess,
+} from "@/lib/integrations/status";
 
 export function getWithingsWebhookCallbackUrl(): string {
   const baseUrl = `${process.env.NEXT_PUBLIC_APP_URL}/api/withings/webhook`;
@@ -48,6 +53,16 @@ export async function getValidToken(userId: string): Promise<{
         getEvent()?.addWarning(
           `No credentials found for user ${userId} during token refresh`,
         );
+        // Without credentials we cannot refresh OR re-authenticate — the
+        // user has to re-enter the client_id/secret. Mark as reauth so
+        // scheduled syncs back off.
+        await recordSyncFailure({
+          userId,
+          integration: "withings",
+          kind: "reauth_required",
+          message: "Withings credentials missing — token refresh skipped",
+          errorCode: "credentials_missing",
+        });
         return null;
       }
 
@@ -72,6 +87,20 @@ export async function getValidToken(userId: string): Promise<{
       };
     } catch (err) {
       getEvent()?.addWarning(`Token refresh failed for user ${userId}: ${err}`);
+      // Withings refresh failures expose the upstream `status` code in
+      // the error message (see refreshAccessToken). We treat any of the
+      // documented permanent-revoke statuses (100, 101, 102, 200..299
+      // for invalid_grant) as reauth-required and everything else as a
+      // transient retryable failure.
+      const message = err instanceof Error ? err.message : String(err);
+      const isReauth = isWithingsRefreshReauthFailure(message);
+      await recordSyncFailure({
+        userId,
+        integration: "withings",
+        kind: isReauth ? "reauth_required" : "transient",
+        message,
+        errorCode: extractWithingsStatus(message),
+      });
       return null;
     }
   }
@@ -88,12 +117,34 @@ export async function getValidToken(userId: string): Promise<{
 /**
  * Sync measurements from Withings for a given user.
  * Fetches data since last sync (or last 30 days if first sync).
+ *
+ * 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<number> {
+  // Park: if the last burst said reauth_required, do nothing until
+  // the OAuth callback flips state back to "connected". Returning 0
+  // matches the existing contract for "no-op sync".
+  if (await isReauthRequired(userId, "withings")) {
+    getEvent()?.addWarning(
+      `withings sync skipped for ${userId}: parked at error_reauth`,
+    );
+    return 0;
+  }
+
   const tokenInfo = await getValidToken(userId);
+  // Note: getValidToken already calls recordSyncFailure on the refresh
+  // path, so we don't double-record here.
   if (!tokenInfo) return 0;
 
   const connection = await prisma.withingsConnection.findUnique({
@@ -107,7 +158,22 @@ export async function syncUserMeasurements(
       ? new Date(connection.lastSyncedAt.getTime() - 60 * 1000) // overlap 1 min to avoid gaps
       : new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); // 30 days back
 
-  const measures = await fetchMeasurements(tokenInfo.accessToken, startDate);
+  let measures: Awaited<ReturnType<typeof fetchMeasurements>>;
+  try {
+    measures = await fetchMeasurements(tokenInfo.accessToken, startDate);
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    await recordSyncFailure({
+      userId,
+      integration: "withings",
+      kind: isWithingsRefreshReauthFailure(message)
+        ? "reauth_required"
+        : "transient",
+      message,
+      errorCode: extractWithingsStatus(message),
+    });
+    throw err;
+  }
 
   let imported = 0;
 
@@ -146,9 +212,48 @@ export async function syncUserMeasurements(
     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");
+
   return imported;
 }
 
+/**
+ * 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>".
+ */
+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;
+}
+
+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];
+}
+
 /**
  * Subscribe to Withings webhook for a user.
  */