fix(status): expose background failures

This commit is contained in:
Marc-André Bombeck committed 2026-07-28 14:00:44 +02:00
1 parent a689329e79
commit 8ece6f39cc
31 files changed
+854 -44

No files matched your search

@@ -44,10 +44,16 @@ vi.mock("@/lib/jobs/worker-status", () => ({
})),
}));
vi.mock("@/lib/jobs/job-failures", () => ({
JOB_FAILURE_WINDOW_HOURS: 72,
readFailingQueues: vi.fn(),
}));
// v1.4.25 W21 Fix-K — `/api/admin/status` now reads the legacy-form
import { GET } from "../route";
import { prisma } from "@/lib/db";
import { requireAdmin, HttpError } from "@/lib/api-handler";
import { readFailingQueues } from "@/lib/jobs/job-failures";
const ADMIN_CTX = {
authMethod: "cookie" as const,
@@ -69,6 +75,7 @@ beforeEach(() => {
vi.mocked(prisma.apiToken.count).mockResolvedValue(0);
vi.mocked(prisma.session.count).mockResolvedValue(0);
vi.mocked(prisma.appSettings.findUnique).mockResolvedValue(null);
vi.mocked(readFailingQueues).mockResolvedValue([]);
});
describe("GET /api/admin/status (integration health summary)", () => {
@@ -148,7 +155,7 @@ describe("GET /api/admin/status (integration health summary)", () => {
const body = (await res.json()) as {
data: {
counts: Record<string, number>;
worker: { running: boolean };
worker: { running: boolean; errors?: number };
database: string;
};
};
@@ -161,10 +168,44 @@ describe("GET /api/admin/status (integration health summary)", () => {
activeSessions: 0,
});
expect(body.data.worker.running).toBe(true);
expect(body.data.worker).not.toHaveProperty("errors");
expect(body.data.database).toBe("connected");
// Regression: only NON-revoked tokens are counted as "active".
const tokenCallArgs = vi.mocked(prisma.apiToken.count).mock.calls[0]?.[0];
expect(tokenCallArgs).toEqual({ where: { revoked: false } });
});
it("reports named failing queues through failingJobs only", async () => {
vi.mocked(readFailingQueues).mockResolvedValueOnce([
{
queue: "environment-fetch",
failures: 2,
lastFailedAt: "2026-07-28T12:00:00.000Z",
lastError: "upstream unavailable",
},
]);
const res = await GET();
const body = (await res.json()) as {
data: {
worker: Record<string, unknown>;
failingJobs: {
windowHours: number;
queues: Array<{ queue: string; failures: number }>;
};
};
};
expect(body.data.worker).not.toHaveProperty("errors");
expect(body.data.failingJobs).toEqual({
windowHours: 72,
queues: [
expect.objectContaining({
queue: "environment-fetch",
failures: 2,
}),
],
});
});
});
+9 -1
View File
@@ -75,7 +75,15 @@ export const GET = apiHandler(async () => {
buildTime: BUILD_TIME,
startTime: START_TIME,
database: "connected",
worker: workerStatus,
worker: {
running: workerStatus.running,
startedAt: workerStatus.startedAt,
lastHeartbeat: workerStatus.lastHeartbeat,
lastReminderCheck: workerStatus.lastReminderCheck,
lastWithingsSync: workerStatus.lastWithingsSync,
lastInsightsRun: workerStatus.lastInsightsRun,
jobsProcessed: workerStatus.jobsProcessed,
},
failingJobs:
failingJobs === null
? null
@@ -0,0 +1,48 @@
import { describe, expect, it } from "vitest";
import { renderToStaticMarkup } from "react-dom/server";
import { I18nProvider } from "@/lib/i18n/context";
import { FailingJobsCard } from "../system-status-section";
function render(node: React.ReactNode) {
return renderToStaticMarkup(
<I18nProvider initialLocale="en">{node}</I18nProvider>,
);
}
describe("FailingJobsCard", () => {
it("renders named exhausted-retry failures from failingJobs", () => {
const html = render(
<FailingJobsCard
failingJobs={{
windowHours: 72,
queues: [
{
queue: "environment-fetch",
failures: 2,
lastFailedAt: "2026-07-28T12:00:00.000Z",
lastError: "upstream unavailable",
},
],
}}
/>,
);
expect(html).toContain("environment-fetch");
expect(html).toContain("2 failures");
expect(html).toContain("upstream unavailable");
expect(html).toContain('data-testid="failing-jobs-list"');
});
it("distinguishes an unavailable queue ledger from an empty one", () => {
const unavailable = render(<FailingJobsCard failingJobs={null} />);
const empty = render(
<FailingJobsCard failingJobs={{ windowHours: 72, queues: [] }} />,
);
expect(unavailable).toContain("No job queue to read");
expect(unavailable).not.toContain('data-testid="failing-jobs-none"');
expect(empty).toContain('data-testid="failing-jobs-none"');
expect(empty).toContain("No job has failed in the last 72 hours.");
});
});
-1
View File
@@ -42,7 +42,6 @@ export interface WorkerStatus {
lastWithingsSync: string | null;
lastInsightsRun: string | null;
jobsProcessed: number;
errors: number;
}
/** One queue with exhausted-retry failures, as `/api/admin/status` reports it. */
@@ -59,7 +59,7 @@ const HostMetricsChart = dynamic(
* means quiet queues, which is why the "cannot ask" case renders its own line
* rather than looking identical to health.
*/
function FailingJobsCard({
export function FailingJobsCard({
failingJobs,
}: {
failingJobs: { windowHours: number; queues: FailingQueue[] } | null;
@@ -217,6 +217,33 @@ describe("IntegrationStatusPill — a failing provider states its age", () => {
expect(html).toContain('data-state="warning"');
});
it("renders the localized failure count against the server threshold", () => {
const html = render(
<IntegrationStatusPill
state="warning"
lastSyncAt={new Date()}
now={new Date()}
failureCount={2}
failureThreshold={3}
/>,
);
expect(html).toContain("2/3");
});
it("omits a partial failure ratio", () => {
const html = render(
<IntegrationStatusPill
state="warning"
lastSyncAt={new Date()}
now={new Date()}
failureCount={2}
/>,
);
expect(html).not.toContain("2/");
});
it("keeps the German label free of a connection claim too", () => {
const html = render(
<IntegrationStatusPill
@@ -69,6 +69,10 @@ interface IntegrationStatusPillProps {
* suffix (used when the integration was just configured but has
* never synced yet). */
lastSyncAt: Date | string | null;
/** Current consecutive-failure count from the active server bucket. */
failureCount?: number;
/** Server-owned alert threshold paired with `failureCount`. */
failureThreshold?: number;
/** Override "now" for deterministic testing. Defaults to `new Date()`. */
now?: Date;
/** Override the testid so a card can keep its own established e2e hook. */
@@ -129,6 +133,8 @@ const RELATIVE_STATES = new Set<IntegrationPillState>([
export function IntegrationStatusPill({
state,
lastSyncAt,
failureCount,
failureThreshold,
now,
testId,
className,
@@ -189,6 +195,15 @@ export function IntegrationStatusPill({
t,
)
: null;
const failureRatio =
failureCount !== undefined &&
failureCount > 0 &&
failureThreshold !== undefined
? t("settings.integrationPill.failureCount", {
count: failureCount,
threshold: failureThreshold,
})
: null;
return (
<Badge
@@ -218,6 +233,7 @@ export function IntegrationStatusPill({
{icon}
<span className="truncate">
{label}
{failureRatio ? ` · ${failureRatio}` : null}
{relative ? <span aria-hidden="true"> · {relative}</span> : null}
</span>
{relative ? <span className="sr-only">{relative}</span> : null}
@@ -13,7 +13,11 @@ import { join, sep } from "node:path";
import { globSync } from "node:fs";
import type { SyncVerdict } from "@/lib/integrations/sync-verdict";
import { pillStateForVerdict } from "../shared";
import {
pickStatus,
pillFailurePropsFor,
pillStateForVerdict,
} from "../shared";
describe("pillStateForVerdict — the verdict→pill table", () => {
const table: Array<[SyncVerdict | undefined, string]> = [
@@ -50,6 +54,61 @@ describe("pillStateForVerdict — the verdict→pill table", () => {
});
});
describe("pillFailurePropsFor", () => {
it("projects the active bucket and envelope threshold onto the pill", () => {
const status = pickStatus(
{
threshold: 3,
integrations: [
{
integration: "withings",
state: "error_transient",
lastSuccessAt: null,
lastAttemptAt: "2026-07-28T12:00:00.000Z",
lastError: "upstream unavailable",
consecutiveFailuresByKind: {
transient: 2,
reauth_required: 0,
persistent: 1,
},
},
],
},
"withings",
);
expect(pillFailurePropsFor(status)).toEqual({
failureCount: 2,
failureThreshold: 3,
});
});
it("returns no ratio for a healthy zero-count status", () => {
const status = pickStatus(
{
threshold: 3,
integrations: [
{
integration: "withings",
state: "connected",
lastSuccessAt: "2026-07-28T12:00:00.000Z",
lastAttemptAt: "2026-07-28T12:00:00.000Z",
lastError: null,
consecutiveFailuresByKind: {
transient: 0,
reauth_required: 0,
persistent: 0,
},
},
],
},
"withings",
);
expect(pillFailurePropsFor(status)).toEqual({});
});
});
describe("structural guard — no second pill dialect", () => {
const SRC = join(process.cwd(), "src");
@@ -185,6 +185,8 @@ function toOAuthStatus(
lastAttemptAt: vm.lastAttemptAt,
lastError: vm.lastError,
legacyLastSyncedAt: vm.legacyLastSyncedAt,
consecutiveFailuresByKind: vm.consecutiveFailuresByKind,
failureThreshold: vm.failureThreshold,
syncHealth: vm.syncHealth,
metricFreshness: vm.metricFreshness,
};
@@ -310,11 +312,16 @@ export function ConnectionsPanel() {
pickStatus(integrationStatus, "strava"),
);
const nightscoutViewModel = pickStatus(integrationStatus, "nightscout");
const unclaimedIntegrations = (integrationStatus?.integrations ?? []).filter(
(entry) =>
!CARD_BACKED_INTEGRATIONS.has(entry.integration) &&
(entry.configured || entry.connected),
);
const unclaimedIntegrations = (integrationStatus?.integrations ?? [])
.filter(
(entry) =>
!CARD_BACKED_INTEGRATIONS.has(entry.integration) &&
(entry.configured || entry.connected),
)
.map((entry) => ({
...entry,
failureThreshold: integrationStatus?.threshold,
}));
return (
<div className="space-y-6" data-slot="connections-panel">
@@ -52,6 +52,7 @@ import {
import {
IntegrationErrorMessage,
pillFailurePropsFor,
pillStateFor,
pillTimestampFor,
type IntegrationStatusViewModel,
@@ -237,6 +238,7 @@ export function FitbitCard({
<IntegrationStatusPill
state={pillState}
lastSyncAt={pillLastSyncAt}
{...pillFailurePropsFor(viewModel)}
/>
}
/>
@@ -60,6 +60,7 @@ import {
import {
IntegrationErrorMessage,
pillFailurePropsFor,
pillStateFor,
pillTimestampFor,
type IntegrationStatusViewModel,
@@ -245,6 +246,7 @@ export function GoogleHealthCard({
<IntegrationStatusPill
state={pillState}
lastSyncAt={pillLastSyncAt}
{...pillFailurePropsFor(viewModel)}
/>
}
/>
@@ -29,6 +29,7 @@ import {
import { useTranslations } from "@/lib/i18n/context";
import {
pillFailurePropsFor,
pillStateFor,
pillTimestampFor,
type IntegrationKey,
@@ -89,6 +90,7 @@ export function IntegrationFallbackRow({
<IntegrationStatusPill
state={pillState}
lastSyncAt={pillTimestampFor(status)}
{...pillFailurePropsFor(status)}
now={now}
/>
}
@@ -59,6 +59,7 @@ import {
} from "./sync-outcome";
import {
IntegrationErrorMessage,
pillFailurePropsFor,
pillStateForVerdict,
pillTimestampFor,
type IntegrationStatusViewModel,
@@ -210,6 +211,7 @@ export function NightscoutCard({
<IntegrationStatusPill
state={pillState}
lastSyncAt={pillTimestampFor(status)}
{...pillFailurePropsFor(status)}
/>
}
/>
@@ -58,7 +58,11 @@ import {
useSyncOutcomeMessage,
type SyncOutcomeState,
} from "./sync-outcome";
import { IntegrationErrorMessage, pillStateForVerdict } from "./shared";
import {
IntegrationErrorMessage,
pillFailurePropsFor,
pillStateForVerdict,
} from "./shared";
import {
IntegrationCardDescription,
IntegrationRedirectGuide,
@@ -81,6 +85,12 @@ export interface OAuthProviderStatus {
lastAttemptAt?: string | null;
lastError?: string | null;
legacyLastSyncedAt?: string | null;
consecutiveFailuresByKind?: {
transient: number;
reauth_required: number;
persistent: number;
} | null;
failureThreshold?: number;
/** The server-resolved liveness verdict; the pill is a projection of it. */
syncHealth?: SyncHealth;
metricFreshness?: MetricFreshnessEntry[];
@@ -270,6 +280,7 @@ export function OAuthProviderCard({
? (status?.syncHealth?.since ?? null)
: (status?.legacyLastSyncedAt ?? status?.lastSuccessAt ?? null)
}
{...pillFailurePropsFor(status)}
/>
}
/>
@@ -50,6 +50,8 @@ export interface IntegrationStatusViewModel {
reauth_required: number;
persistent: number;
} | null;
/** Client-side copy of the envelope threshold, attached by `pickStatus`. */
failureThreshold?: number;
configured?: boolean;
connected?: boolean;
connectedAt?: string | null;
@@ -110,7 +112,12 @@ export function pickStatus(
envelope: IntegrationStatusEnvelope | undefined,
integration: IntegrationKey,
): IntegrationStatusViewModel | undefined {
return envelope?.integrations.find((i) => i.integration === integration);
const status = envelope?.integrations.find(
(entry) => entry.integration === integration,
);
return status
? { ...status, failureThreshold: envelope?.threshold }
: undefined;
}
/**
@@ -158,6 +165,36 @@ export function pillStateFor(
return pillStateForVerdict(status?.syncHealth?.verdict);
}
export interface IntegrationPillFailureProps {
failureCount?: number;
failureThreshold?: number;
}
/**
* Pair the largest per-kind streak with the server-owned alert threshold. The
* alert ladder uses the same maximum, so the ratio cannot disagree with when a
* notification becomes eligible.
*/
export function pillFailurePropsFor(
status:
| Pick<
IntegrationStatusViewModel,
"consecutiveFailuresByKind" | "failureThreshold"
>
| undefined,
): IntegrationPillFailureProps {
const buckets = status?.consecutiveFailuresByKind;
const failureThreshold = status?.failureThreshold;
if (!buckets || failureThreshold === undefined) return {};
const failureCount = Math.max(
buckets.transient,
buckets.reauth_required,
buckets.persistent,
);
return failureCount > 0 ? { failureCount, failureThreshold } : {};
}
/**
* The timestamp the pill shows inline. `connected` reads the last success (a
* connection row's own `lastSyncedAt` wins when it is newer, which is what
@@ -43,6 +43,7 @@ import {
import {
IntegrationErrorMessage,
pillFailurePropsFor,
pillStateFor,
pillTimestampFor,
type IntegrationStatusViewModel,
@@ -218,6 +219,7 @@ export function WhoopCard({
<IntegrationStatusPill
state={pillState}
lastSyncAt={pillLastSyncAt}
{...pillFailurePropsFor(viewModel)}
/>
}
/>
@@ -43,6 +43,7 @@ import {
import {
IntegrationErrorMessage,
pillStateFor,
pillFailurePropsFor,
pillTimestampFor,
type IntegrationStatusViewModel,
} from "./shared";
@@ -230,6 +231,7 @@ export function WithingsCard({
<IntegrationStatusPill
state={pillState}
lastSyncAt={pillLastSyncAt}
{...pillFailurePropsFor(viewModel)}
/>
}
/>
@@ -36,7 +36,11 @@ vi.mock("@/lib/auth/audit", () => ({
}));
vi.mock("@/lib/notifications/dispatcher", () => ({
dispatchNotification: vi.fn().mockResolvedValue(undefined),
dispatchNotification: vi.fn().mockResolvedValue({
dispatched: true,
channelsAttempted: 1,
channelsSucceeded: 1,
}),
}));
vi.mock("@/lib/crypto", () => ({
+105 -6
View File
@@ -9,6 +9,15 @@
*/
import { describe, it, expect, vi, beforeEach } from "vitest";
const { addWarningMock } = vi.hoisted(() => ({
addWarningMock: vi.fn(),
}));
vi.mock("@/lib/logging/context", () => ({
annotate: vi.fn(),
getEvent: () => ({ addWarning: addWarningMock }),
}));
vi.mock("@/lib/db", () => ({
prisma: {
integrationStatus: {
@@ -28,7 +37,7 @@ vi.mock("@/lib/auth/audit", () => ({
}));
vi.mock("@/lib/notifications/dispatcher", () => ({
dispatchNotification: vi.fn().mockResolvedValue(undefined),
dispatchNotification: vi.fn(),
}));
vi.mock("@/lib/crypto", () => ({
@@ -56,6 +65,11 @@ import { dispatchNotification } from "@/lib/notifications/dispatcher";
beforeEach(() => {
vi.resetAllMocks();
vi.mocked(dispatchNotification).mockResolvedValue({
dispatched: true,
channelsAttempted: 1,
channelsSucceeded: 1,
});
delete process.env.INTEGRATION_FAILURE_ALERT_THRESHOLD;
});
@@ -207,6 +221,90 @@ describe("recordSyncFailure — at threshold", () => {
);
});
it("does not stamp alertedAt when no admin channel is available", async () => {
vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({
consecutiveFailuresByKind: {
transient: 2,
reauth_required: 0,
persistent: 0,
},
persistentFailureStartedAt: null,
alertedAt: null,
} as never);
vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({
alertedAt: null,
} as never);
vi.mocked(prisma.user.findUnique).mockResolvedValueOnce({
email: "user@example.com",
} as never);
vi.mocked(prisma.user.findMany).mockResolvedValueOnce([
{ id: "admin-1" },
] as never);
vi.mocked(dispatchNotification).mockResolvedValueOnce({
dispatched: false,
channelsAttempted: 0,
channelsSucceeded: 0,
});
await recordSyncFailure({
userId: "u1",
integration: "withings",
kind: "transient",
message: "Withings refresh error: 503 - upstream service down",
errorCode: "503",
});
expect(dispatchNotification).toHaveBeenCalledOnce();
expect(prisma.integrationStatus.update).not.toHaveBeenCalled();
expect(addWarningMock).toHaveBeenCalledWith(
"No admin notification channel available for withings failure alert",
);
});
it("stamps alertedAt after an attempted delivery that failed", async () => {
vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({
consecutiveFailuresByKind: {
transient: 2,
reauth_required: 0,
persistent: 0,
},
persistentFailureStartedAt: null,
alertedAt: null,
} as never);
vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({
alertedAt: null,
} as never);
vi.mocked(prisma.user.findUnique).mockResolvedValueOnce({
email: "user@example.com",
} as never);
vi.mocked(prisma.user.findMany).mockResolvedValueOnce([
{ id: "admin-1" },
] as never);
vi.mocked(dispatchNotification).mockResolvedValueOnce({
dispatched: false,
channelsAttempted: 1,
channelsSucceeded: 0,
});
vi.mocked(prisma.integrationStatus.update).mockResolvedValueOnce(
{} as never,
);
await recordSyncFailure({
userId: "u1",
integration: "withings",
kind: "transient",
message: "Withings refresh error: 503 - upstream service down",
errorCode: "503",
});
expect(prisma.integrationStatus.update).toHaveBeenCalledWith({
where: {
userId_integration: { userId: "u1", integration: "withings" },
},
data: { alertedAt: expect.any(Date) },
});
});
it("uses the custom threshold from INTEGRATION_FAILURE_ALERT_THRESHOLD", async () => {
process.env.INTEGRATION_FAILURE_ALERT_THRESHOLD = "5";
// 4 < 5 → should not page yet. Seed the existing bucket at 3 so
@@ -398,8 +496,8 @@ describe("recordSyncFailure — persistent (v1.4.42 W6)", () => {
});
});
describe("recordSyncFailure — admin Telegram skipped silently when no admins", () => {
it("logs a wide-event warning but does NOT throw when no admin users exist", async () => {
describe("recordSyncFailure — no admin recipients", () => {
it("surfaces the unavailable delivery and leaves alertedAt unstamped", async () => {
// Seed bucket at 2 so the in-memory increment hits the threshold
// and the no-admins branch is the one being exercised.
vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({
@@ -418,9 +516,6 @@ describe("recordSyncFailure — admin Telegram skipped silently when no admins",
email: "u@example.com",
} as never);
vi.mocked(prisma.user.findMany).mockResolvedValueOnce([] as never);
vi.mocked(prisma.integrationStatus.update).mockResolvedValueOnce(
{} as never,
);
await expect(
recordSyncFailure({
@@ -432,6 +527,10 @@ describe("recordSyncFailure — admin Telegram skipped silently when no admins",
).resolves.toBeUndefined();
expect(dispatchNotification).not.toHaveBeenCalled();
expect(prisma.integrationStatus.update).not.toHaveBeenCalled();
expect(addWarningMock).toHaveBeenCalledWith(
"No admin notification channel available for withings failure alert",
);
});
});
+46 -24
View File
@@ -35,7 +35,10 @@ import { prisma } from "@/lib/db";
import { encrypt, decrypt } from "@/lib/crypto";
import { auditLog } from "@/lib/auth/audit";
import { annotate, getEvent } from "@/lib/logging/context";
import { dispatchNotification } from "@/lib/notifications/dispatcher";
import {
dispatchNotification,
type DispatchOutcome,
} from "@/lib/notifications/dispatcher";
import type { IntegrationClassification } from "@/lib/integrations/http-status-classifier";
export type IntegrationKey =
@@ -755,7 +758,7 @@ export async function recordSyncFailure(
row.alertedAt &&
now.getTime() - row.alertedAt.getTime() < ALERT_REPEAT_WINDOW_MS;
if (!previouslyAlerted) {
await maybeAlertAdmins({
const outcome = await maybeAlertAdmins({
userId,
integration,
kind: effectiveKind,
@@ -766,14 +769,25 @@ export async function recordSyncFailure(
getEvent()?.addWarning(
`Admin alert dispatch failed: ${err instanceof Error ? err.message : String(err)}`,
);
return null;
});
// Stamp the alert window even if dispatch failed — better to
// miss one notification than to flood admins on a backend
// outage.
await prisma.integrationStatus.update({
where: { userId_integration: { userId, integration } },
data: { alertedAt: now },
});
if (outcome?.channelsAttempted) {
// An attempted channel owns its own retry/cooldown policy. Stamp the
// alert window even when every attempt failed so a backend outage
// cannot turn this failure streak into an hourly notification flood.
await prisma.integrationStatus.update({
where: { userId_integration: { userId, integration } },
data: { alertedAt: now },
});
} else if (outcome) {
// No recipient channel accepted an attempt. Leaving alertedAt empty
// keeps the missing delivery visible instead of recording a page that
// never left the process.
getEvent()?.addWarning(
`No admin notification channel available for ${integration} failure alert`,
);
}
}
}
@@ -1203,18 +1217,15 @@ export function formatAdminAlertPayload(input: AlertInput): {
}
/**
* Page admins on Telegram via the existing dispatcher. We do NOT add
* a new sender, channel type, or retry path — B3 owns that surface.
* The dispatcher is opt-in per-event and per-channel, so an admin
* who has not configured Telegram simply gets no message (silent
* fall-through is the documented behaviour).
* Page admins through the existing dispatcher. We do NOT add a new sender,
* channel type, or retry path. The dispatcher is opt-in per event and channel,
* and its outcome is the authority on whether a delivery was attempted.
*
* The notification is sent to EACH admin user with a Telegram channel
* configured. We resolve recipients once per failure burst (gated by
* `alertedAt` upstream) so a deployment with a single admin doesn't
* pay for a query per user.
* The notification is sent to each admin user. The returned outcome aggregates
* the dispatcher's real per-admin outcomes so one attempted delivery cannot be
* hidden by another admin having no enabled channel.
*/
async function maybeAlertAdmins(input: AlertInput): Promise<void> {
async function maybeAlertAdmins(input: AlertInput): Promise<DispatchOutcome> {
const subject = await prisma.user.findUnique({
where: { id: input.userId },
select: { email: true },
@@ -1226,24 +1237,35 @@ async function maybeAlertAdmins(input: AlertInput): Promise<void> {
});
if (admins.length === 0) {
getEvent()?.addWarning(
`No admin user found to alert about persistent ${input.integration} failure for user ${input.userId}`,
);
return;
return {
dispatched: false,
channelsAttempted: 0,
channelsSucceeded: 0,
};
}
const payload = formatAdminAlertPayload({
...input,
subjectLabel: subject?.email ?? input.userId,
});
const outcome: DispatchOutcome = {
dispatched: false,
channelsAttempted: 0,
channelsSucceeded: 0,
};
for (const admin of admins) {
await dispatchNotification({
const adminOutcome = await dispatchNotification({
eventType: "SYSTEM_ALERT",
userId: admin.id,
title: payload.title,
message: payload.message,
metadata: payload.metadata,
});
outcome.dispatched ||= adminOutcome.dispatched;
outcome.channelsAttempted += adminOutcome.channelsAttempted;
outcome.channelsSucceeded += adminOutcome.channelsSucceeded;
}
return outcome;
}
@@ -0,0 +1,29 @@
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { describe, expect, it } from "vitest";
import {
JOB_FAILURE_WINDOW_HOURS,
PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS,
} from "@/lib/jobs/job-failures";
const PG_BOSS_ROOT = join(process.cwd(), "node_modules/pg-boss");
describe("pg-boss failed-row ledger retention", () => {
it("pins the installed source assumptions behind the 72-hour reader", () => {
const packageJson = JSON.parse(
readFileSync(join(PG_BOSS_ROOT, "package.json"), "utf8"),
) as { version: string };
const plans = readFileSync(join(PG_BOSS_ROOT, "dist/plans.js"), "utf8");
expect(packageJson.version).toBe("12.26.0");
expect(plans).toContain("retention_seconds: FORTEEN_DAYS");
expect(plans).toContain("deletion_seconds: SEVEN_DAYS");
expect(plans).toContain(
"completed_on + deletion_seconds * interval '1s' < now()",
);
expect(JOB_FAILURE_WINDOW_HOURS).toBeLessThan(
PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS,
);
});
});
@@ -0,0 +1,151 @@
import { existsSync, readFileSync } from "node:fs";
import { dirname, join, resolve } from "node:path";
import { describe, expect, it } from "vitest";
import {
SUBSYSTEM_SURFACE,
type SubsystemSurface,
} from "@/lib/jobs/subsystem-surface";
const REGISTRARS = [
"register-integration-sync.ts",
"register-status.ts",
"register-rollup.ts",
"register-reminders.ts",
"register-maintenance.ts",
].map((file) => join(process.cwd(), "src/lib/jobs/reminder", file));
function withoutComments(source: string): string {
return source.replace(/\/\*[\s\S]*?\*\//g, "").replace(/\/\/.*$/gm, "");
}
export function extractAllQueueBindings(source: string): string[] {
const match = /const\s+allQueues\s*=\s*\[([\s\S]*?)\n\s*\];/.exec(source);
if (!match) throw new Error("Registrar has no allQueues array");
return withoutComments(match[1])
.split(",")
.map((entry) => entry.trim())
.filter(Boolean)
.map((entry) => {
if (!/^[A-Z][A-Z0-9_]*$/.test(entry)) {
throw new Error(`Unsupported allQueues entry: ${entry}`);
}
return entry;
});
}
function literalPattern(binding: string): RegExp {
return new RegExp(
`(?:export\\s+)?const\\s+${binding}(?:\\s*:[^=;]+)?\\s*=\\s*["']([^"']+)["']`,
);
}
function modulePath(fromFile: string, specifier: string): string {
const base = specifier.startsWith("@/")
? resolve(process.cwd(), "src", specifier.slice(2))
: resolve(dirname(fromFile), specifier);
const candidates = [`${base}.ts`, `${base}.tsx`, join(base, "index.ts")];
const found = candidates.find((candidate) => existsSync(candidate));
if (!found) throw new Error(`Cannot resolve ${specifier} from ${fromFile}`);
return found;
}
function resolveQueueBinding(
binding: string,
file: string,
seen: Set<string> = new Set(),
): string {
const visit = `${file}:${binding}`;
if (seen.has(visit)) throw new Error(`Cyclic queue binding: ${visit}`);
seen.add(visit);
const source = withoutComments(readFileSync(file, "utf8"));
const local = literalPattern(binding).exec(source);
if (local) return local[1];
const imports = /import\s*{([\s\S]*?)}\s*from\s*["']([^"']+)["'];?/g;
for (const match of source.matchAll(imports)) {
for (const rawToken of match[1].split(",")) {
const token = rawToken.trim().replace(/^type\s+/, "");
if (!token) continue;
const [imported, alias] = token
.split(/\s+as\s+/)
.map((part) => part.trim());
if ((alias ?? imported) !== binding) continue;
return resolveQueueBinding(imported, modulePath(file, match[2]), seen);
}
}
throw new Error(`Cannot resolve queue binding ${binding} from ${file}`);
}
function registeredQueueNames(): string[] {
return REGISTRARS.flatMap((file) => {
const source = readFileSync(file, "utf8");
return extractAllQueueBindings(source).map((binding) =>
resolveQueueBinding(binding, file),
);
});
}
export function findSurfaceDrift(
queueNames: readonly string[],
surfaces: Readonly<Record<string, SubsystemSurface>>,
): { missing: string[]; extra: string[] } {
const registered = new Set(queueNames);
const classified = new Set(Object.keys(surfaces));
return {
missing: [...registered].filter((name) => !classified.has(name)).sort(),
extra: [...classified].filter((name) => !registered.has(name)).sort(),
};
}
function assertSurfaceComplete(
queueNames: readonly string[],
surfaces: Readonly<Record<string, SubsystemSurface>>,
): void {
const drift = findSurfaceDrift(queueNames, surfaces);
if (drift.missing.length === 0 && drift.extra.length === 0) return;
throw new Error(
`Subsystem surface drift: missing=[${drift.missing.join(", ")}], extra=[${drift.extra.join(", ")}]`,
);
}
describe("SUBSYSTEM_SURFACE", () => {
it("classifies every current registrar queue with no stale entries", () => {
const names = registeredQueueNames();
expect(new Set(names).size).toBe(names.length);
expect(() => assertSurfaceComplete(names, SUBSYSTEM_SURFACE)).not.toThrow();
});
it("requires none reasons to cite a test path or issue", () => {
for (const surface of Object.values(SUBSYSTEM_SURFACE)) {
if (surface.audience !== "none") continue;
expect(surface.reason).toMatch(/^(?:#\d+|.+\.test\.tsx?(?::\d+)?)$/);
}
});
it("catches a new allQueues binding without a surface entry", () => {
const fixture = `
const EXISTING_QUEUE = "existing-queue";
const NEW_QUEUE = "new-queue";
const allQueues = [
EXISTING_QUEUE,
NEW_QUEUE,
];
`;
const names = extractAllQueueBindings(fixture).map((binding) => {
const match = literalPattern(binding).exec(withoutComments(fixture));
if (!match)
throw new Error(`Fixture binding did not resolve: ${binding}`);
return match[1];
});
expect(() =>
assertSurfaceComplete(names, {
"existing-queue": { audience: "system" },
}),
).toThrow(/missing=\[new-queue\]/);
});
});
+14
View File
@@ -27,6 +27,20 @@
import { prisma } from "@/lib/db";
import { redactSecrets } from "@/lib/logging/redact";
/**
* pg-boss 12.26.0 stores terminal rows until
* `completed_on + deletion_seconds` (`dist/plans.js`). Its queue default is
* 604800 seconds, while `retention_seconds` defaults to 1209600 seconds; job
* insertion copies those queue values and maintenance applies the terminal-row
* deletion clock. The deployed queue-table audit found those same values on
* every queue, and no job submitter overrides either option.
*
* Seven days is therefore the verified availability floor for failed rows.
* The reader stays at 72 hours, safely inside that floor, and pg-boss remains
* the single failure ledger.
*/
export const PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS = 7 * 24;
/** How far back a failure still counts as news. */
export const JOB_FAILURE_WINDOW_HOURS = 72;
+138
View File
@@ -0,0 +1,138 @@
/**
* Primary audience for a terminal queue failure. `account` queues already
* have an affected-account status surface; `system` queues use the admin
* failing-jobs card. An intentional `none` must point to an executable test or
* tracked issue so a prose-only exemption cannot hide a new queue.
*/
export type SubsystemSurface =
| { audience: "account"; reason?: never }
| { audience: "system"; reason?: never }
| {
audience: "none";
reason:
| `#${number}`
| `${string}.test.ts${string}`
| `${string}.test.tsx${string}`;
};
const subsystemSurface = {
// register-integration-sync.ts
"withings-fallback-sync": { audience: "account" },
"withings-activity-sync": { audience: "account" },
"withings-sleep-sync": { audience: "account" },
"withings-ecg-sync": { audience: "account" },
"withings-oauth-state-cleanup": { audience: "system" },
"whoop-recovery-sync": { audience: "account" },
"whoop-sleep-sync": { audience: "account" },
"whoop-workout-sync": { audience: "account" },
"whoop-cycle-sync": { audience: "account" },
"whoop-backfill": { audience: "system" },
"whoop-oauth-state-cleanup": { audience: "system" },
"fitbit-sync": { audience: "account" },
"fitbit-backfill": { audience: "system" },
"fitbit-sleep-repair": { audience: "system" },
"fitbit-oauth-state-cleanup": { audience: "system" },
"google-health-sync": { audience: "account" },
"google-health-backfill": { audience: "system" },
"google-health-sleep-repair": { audience: "system" },
"google-health-oauth-state-cleanup": { audience: "system" },
"sleep-timeline-backfill": { audience: "system" },
"lab-biomarker-backfill": { audience: "system" },
"integration-backfill-admission": { audience: "system" },
"nightscout-sync": { audience: "account" },
"polar-sync": { audience: "account" },
"oura-sync": { audience: "account" },
"strava-sync": { audience: "account" },
"strava-backfill": { audience: "system" },
"oidc-native-handoff-cleanup": { audience: "system" },
// register-status.ts
"insights-general-status": { audience: "system" },
"insights-blood-pressure-status": { audience: "system" },
"insights-weight-status": { audience: "system" },
"insights-pulse-status": { audience: "system" },
"insights-bmi-status": { audience: "system" },
"insights-mood-status": { audience: "system" },
"insights-medication-compliance-status": { audience: "system" },
"insight-pregenerate": { audience: "system" },
"insight-status-generate": { audience: "system" },
"morning-digest-refresh": { audience: "system" },
"data-arrival": { audience: "system" },
"reaction-line-generate": { audience: "system" },
"workout-insight-generate": { audience: "system" },
"recovery-score-compute": { audience: "system" },
"stress-score-compute": { audience: "system" },
"strain-score-compute": { audience: "system" },
"period-narrative-warm": { audience: "system" },
"coach-memory-refresh": { audience: "system" },
"coach-nudge": { audience: "system" },
"coach-reminder-sweep": { audience: "system" },
"coach-plan-review": { audience: "system" },
"medication-low-stock": { audience: "system" },
"daily-briefing": { audience: "system" },
// register-rollup.ts
"rollup-recompute": { audience: "system" },
"rollup-full-backfill": { audience: "system" },
"mood-rollup-recompute": { audience: "system" },
"mood-rollup-full-backfill": { audience: "system" },
"medication-compliance-full-backfill": { audience: "system" },
"step-consolidation": { audience: "system" },
"step-consolidation-repair": { audience: "system" },
"cumulative-pr-rederive": { audience: "system" },
"mean-consolidation": { audience: "system" },
"dense-intraday-retention": { audience: "system" },
"dense-intraday-hourly-rebuild": { audience: "system" },
"drain-per-sample-cumulative": { audience: "system" },
// register-reminders.ts
"medication-reminder-check": { audience: "system" },
"mood-reminder-check": { audience: "system" },
"cycle-reminder-check": { audience: "system" },
"measurement-reminder-check": { audience: "system" },
"reminder-satisfy": { audience: "system" },
// register-maintenance.ts
"environment-fetch": { audience: "account" },
"data-backup": { audience: "system" },
"rate-limit-cleanup": { audience: "system" },
"idempotency-cleanup": { audience: "system" },
"audit-log-cleanup": { audience: "system" },
"step-up-elevation-cleanup": { audience: "system" },
"data-backup-offhost": { audience: "system" },
"data-restore-drill": { audience: "system" },
"host-metric-sample": { audience: "system" },
"feedback-aggregator": { audience: "system" },
"geo-backfill": { audience: "system" },
"tls-pin-monitor": { audience: "system" },
"pr-detection": { audience: "system" },
"medication-inventory-expire": { audience: "system" },
"intake-auto-skip": { audience: "system" },
"apple-health-import-v2": { audience: "account" },
"apple-health-import": { audience: "account" },
"apple-health-import-reconcile": { audience: "account" },
"medication-intake-import": { audience: "system" },
"intake-slot-dedup": { audience: "system" },
"mood-reminder-cleanup": { audience: "system" },
"push-attempt-cleanup": { audience: "system" },
"arrival-reaction-cleanup": { audience: "system" },
"measurement-tombstone-cleanup": { audience: "system" },
"coach-message-cleanup": { audience: "system" },
"note-encryption-backfill": { audience: "system" },
"encryption-key-rotate": { audience: "system" },
"mcp-token-cleanup": { audience: "system" },
"med-notes-encryption-backfill": { audience: "system" },
"document-tombstone-purge": { audience: "system" },
"document-content-index-backfill": { audience: "account" },
"document-index": { audience: "account" },
"document-thumbnail": { audience: "account" },
"document-thumbnail-backfill": { audience: "account" },
"document-summary": { audience: "account" },
"document-summary-catchup": { audience: "account" },
} as const satisfies Record<string, SubsystemSurface>;
/** The queue-name union follows the registrar-guarded registry, never a count. */
export type QueueName = keyof typeof subsystemSurface;
export const SUBSYSTEM_SURFACE: Record<QueueName, SubsystemSurface> =
subsystemSurface;