From 6a3861122c3bfff7b05732da3ff8f85279591f89 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 29 Jul 2026 14:50:30 +0200 Subject: [PATCH] feat(01-22): publish truthful cohort outcomes - fold complete partial failed parked and skipped user verdicts - bound serialized job facts to a privacy-safe scalar allowlist - preserve provider retry isolation and reminder queue completion --- src/lib/jobs/job-outcome.ts | 116 ++++++++++- src/lib/jobs/reminder/google-health-sync.ts | 93 +++++---- src/lib/jobs/reminder/poll-cohort.ts | 131 +++++++++++- src/lib/jobs/reminder/withings-sync.ts | 211 +++++++++++--------- 4 files changed, 410 insertions(+), 141 deletions(-) diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 34b4c8b8e..34db35a3e 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -25,7 +25,7 @@ * what makes them visible at all — today they are visible nowhere. * * `did` is the small pinned-shape bag of facts about the run — counts, - * identifiers, booleans. It rides into the failure report and, once the + * stable codes, booleans. It rides into the failure report and, once the * failure ledger lands, into the operator surface. Keep it to scalars that * a dashboard can key on; it is not a place for payload data or free text. */ @@ -36,6 +36,45 @@ export type JobFact = string | number | boolean; /** The pinned-shape bag of facts a handler reports about its run. */ export type JobFacts = Readonly>; +/** + * Facts cross a persistence/observability boundary. Keep their vocabulary + * explicit so a caller cannot attach an identifier, credential, URL, + * free-form error, or health value to a durable job result. + */ +export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ + "provider", + "outcome", + "total", + "users_synced", + "users_complete", + "users_partial", + "users_failed", + "users_parked", + "users_skipped", + "users_useful", + "users_clean_zero", + "users_retryable", + "downstream_failed", + "measurements_imported", + "deleted", + "generated", + "jobs", + "skipped", + "processed", + "created", + "updated", + "failed", + "retried", + "queued", + "sent", + "subscription_repair_failed", +]); + +export const MAX_JOB_FACTS = 32; +export const MAX_JOB_OUTCOME_BYTES = 2_048; + +const STABLE_CODE_PATTERN = /^[a-z][a-z0-9_]{0,63}$/; + /** * What a handler did. `ok: true` means the work named by the queue actually * happened; `ok: false` means it did not, and `runJob` will fail the pg-boss @@ -79,3 +118,78 @@ export function jobFailed( ): JobOutcome { return { ok: false, reason, cause, did }; } + +export type SerializedJobOutcome = + | { readonly ok: true; readonly did: JobFacts } + | { + readonly ok: false; + readonly reason_code: string; + readonly did?: JobFacts; + }; + +function assertStableCode(value: string, label: string): void { + if (!STABLE_CODE_PATTERN.test(value)) { + throw new TypeError(`${label} must be a stable lowercase code`); + } +} + +function validateFacts(facts: JobFacts): JobFacts { + const entries = Object.entries(facts); + + if (entries.length > MAX_JOB_FACTS) { + throw new RangeError(`job outcome exceeds ${MAX_JOB_FACTS} facts`); + } + + for (const [key, value] of entries) { + if (!JOB_FACT_ALLOWLIST.has(key)) { + throw new TypeError(`job fact key is not allowed: ${key}`); + } + + if (typeof value === "number") { + if (!Number.isSafeInteger(value) || value < 0) { + throw new TypeError( + `job fact ${key} must be a non-negative safe integer`, + ); + } + continue; + } + + if (typeof value === "string") { + assertStableCode(value, `job fact ${key}`); + continue; + } + + if (typeof value !== "boolean") { + throw new TypeError(`job fact ${key} must be a bounded scalar`); + } + } + + return Object.freeze(Object.fromEntries(entries)) as JobFacts; +} + +/** + * Produce the only representation suitable for persistence or structured + * logging. The raw `cause` deliberately remains in-process for retry handling. + */ +export function serializeJobOutcome(outcome: JobOutcome): SerializedJobOutcome { + const did = validateFacts(outcome.did ?? {}); + const serialized: SerializedJobOutcome = outcome.ok + ? { ok: true, did } + : (() => { + assertStableCode(outcome.reason, "job failure reason"); + return outcome.did === undefined + ? { ok: false, reason_code: outcome.reason } + : { ok: false, reason_code: outcome.reason, did }; + })(); + + const byteLength = new TextEncoder().encode( + JSON.stringify(serialized), + ).length; + if (byteLength > MAX_JOB_OUTCOME_BYTES) { + throw new RangeError( + `serialized job outcome exceeds ${MAX_JOB_OUTCOME_BYTES} bytes`, + ); + } + + return serialized; +} diff --git a/src/lib/jobs/reminder/google-health-sync.ts b/src/lib/jobs/reminder/google-health-sync.ts index d6791814b..ac389745a 100644 --- a/src/lib/jobs/reminder/google-health-sync.ts +++ b/src/lib/jobs/reminder/google-health-sync.ts @@ -5,13 +5,19 @@ * queue names, cron schedules, and boss.work registrations. */ import { type Job } from "pg-boss"; +import pLimit from "p-limit"; import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError } from "@/lib/jobs/worker-status"; import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; -import { runGoogleHealthPollCohort } from "@/lib/google-health/sync"; +import { syncUserGoogleHealth } from "@/lib/google-health/sync"; +import { isReauthRequired } from "@/lib/integrations/status"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup"; +import { + foldIntegrationCohortOutcomes, + type IntegrationUserVerdict, +} from "./poll-cohort"; import { getWorkerPrisma } from "./shared"; /** @@ -42,56 +48,69 @@ export async function handleGoogleHealthSync( targets.push(...connections); } if (targets.length === 0) { - return jobDone({ - total: 0, - users_synced: 0, - users_failed: 0, - measurements_imported: 0, + return foldIntegrationCohortOutcomes({ + provider: "google_health", + verdicts: [], }); } // 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. - let usersFailed = 0; - const { usersSynced, measurementsImported } = - await runGoogleHealthPollCohort( - targets.map((t) => t.userId), - { - onUserError: (userId, err) => { - usersFailed++; - 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) { + const limit = pLimit(4); + const verdicts = await Promise.all( + targets.map(({ userId }) => + limit(async (): Promise => { + if (await isReauthRequired(userId, "google-health")) { + return { status: "parked", imported: 0 }; + } + + try { + const result = await syncUserGoogleHealth(userId); + const parked = + result.failed && + result.imported === 0 && + (await isReauthRequired(userId, "google-health")); + const status = parked + ? "parked" + : result.failed + ? result.imported > 0 + ? "partial" + : "failed" + : "complete"; + + if (result.imported > 0) { fireAndForget(enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } - }, - }, - ); + + return { + status, + imported: result.imported, + retryable: result.failed && !parked, + }; + } catch { + evt.addWarning( + "job.google_health_sync failed for one cohort member", + ); + return { status: "failed", imported: 0, retryable: true }; + } + }), + ), + ); + const outcome = foldIntegrationCohortOutcomes({ + provider: "google_health", + verdicts, + }); evt.setBackground({ task_name: "job.google_health_sync", - result: { - users_synced: usersSynced, - total: targets.length, - measurements_imported: measurementsImported, - }, + result: outcome.did, }); // The pass is the unit of success: a revoked grant belongs on that user's // integration ledger, not in a queue failure that retries every account. - return jobDone({ - total: targets.length, - users_synced: usersSynced, - users_failed: usersFailed, - measurements_imported: measurementsImported, - }); + return outcome; } catch (err) { evt.setError(err); recordError(); @@ -117,8 +136,8 @@ export async function handleGoogleHealthOAuthStateCleanup( evt.addMeta("google_health_oauth_state_cleanup_deleted", deleted); return jobDone({ deleted }); } catch (err) { - evt.addWarning(`google-health-oauth-state-cleanup failed: ${err}`); - return jobFailed("google health oauth state cleanup failed", err); + evt.addWarning("google-health-oauth-state-cleanup failed"); + return jobFailed("google_health_oauth_state_cleanup_failed", err); } }, ); diff --git a/src/lib/jobs/reminder/poll-cohort.ts b/src/lib/jobs/reminder/poll-cohort.ts index 154f61af0..7bd205f59 100644 --- a/src/lib/jobs/reminder/poll-cohort.ts +++ b/src/lib/jobs/reminder/poll-cohort.ts @@ -27,6 +27,129 @@ import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { isSyncFailureRecorded } from "@/lib/integrations/status"; import { getWorkerPrisma } from "./shared"; +export type IntegrationCohortProvider = "google_health" | "withings"; + +export type IntegrationUserVerdict = Readonly<{ + status: "complete" | "partial" | "failed" | "parked" | "skipped"; + imported: number; + downstreamFailed?: boolean; + retryable?: boolean; + reasonCode?: string; +}>; + +export type IntegrationCohortFacts = Readonly<{ + provider: IntegrationCohortProvider; + outcome: "useful" | "clean_zero"; + total: number; + users_synced: number; + users_complete: number; + users_partial: number; + users_failed: number; + users_parked: number; + users_skipped: number; + users_useful: number; + users_clean_zero: number; + users_retryable: number; + downstream_failed: number; + measurements_imported: number; +}>; + +export function foldIntegrationCohortOutcomes(input: { + provider: IntegrationCohortProvider; + verdicts: readonly IntegrationUserVerdict[]; + downstreamFailures?: number; +}): { readonly ok: true; readonly did: IntegrationCohortFacts } { + let usersComplete = 0; + let usersPartial = 0; + let usersFailed = 0; + let usersParked = 0; + let usersSkipped = 0; + let usersUseful = 0; + let usersCleanZero = 0; + let usersRetryable = 0; + if ( + input.downstreamFailures !== undefined && + (!Number.isSafeInteger(input.downstreamFailures) || + input.downstreamFailures < 0) + ) { + throw new TypeError( + "cohort downstream failure count must be a non-negative safe integer", + ); + } + let downstreamFailed = input.downstreamFailures ?? 0; + let measurementsImported = 0; + + for (const verdict of input.verdicts) { + if (!Number.isSafeInteger(verdict.imported) || verdict.imported < 0) { + throw new TypeError( + "cohort imported count must be a non-negative safe integer", + ); + } + + switch (verdict.status) { + case "complete": + usersComplete++; + if (verdict.imported === 0) usersCleanZero++; + break; + case "partial": + usersPartial++; + break; + case "failed": + usersFailed++; + break; + case "parked": + usersParked++; + break; + case "skipped": + usersSkipped++; + break; + default: { + const exhaustive: never = verdict.status; + throw new TypeError(`unsupported cohort status: ${String(exhaustive)}`); + } + } + + if ( + (verdict.status === "complete" || verdict.status === "partial") && + verdict.imported > 0 + ) { + usersUseful++; + } + if (verdict.retryable === true) usersRetryable++; + if (verdict.downstreamFailed === true) { + if (!Number.isSafeInteger(downstreamFailed + 1)) { + throw new RangeError( + "cohort downstream failure total exceeds safe integer range", + ); + } + downstreamFailed++; + } + + const nextImported = measurementsImported + verdict.imported; + if (!Number.isSafeInteger(nextImported)) { + throw new RangeError("cohort imported total exceeds safe integer range"); + } + measurementsImported = nextImported; + } + + return jobDone({ + provider: input.provider, + outcome: usersUseful > 0 ? "useful" : "clean_zero", + total: input.verdicts.length, + users_synced: usersComplete, + users_complete: usersComplete, + users_partial: usersPartial, + users_failed: usersFailed, + users_parked: usersParked, + users_skipped: usersSkipped, + users_useful: usersUseful, + users_clean_zero: usersCleanZero, + users_retryable: usersRetryable, + downstream_failed: downstreamFailed, + measurements_imported: measurementsImported, + }) as { readonly ok: true; readonly did: IntegrationCohortFacts }; +} + /** Every poll-cohort handler accepts an optional single-user payload. */ export interface PollCohortPayload { userId?: string; @@ -113,14 +236,12 @@ export function makePollCohortHandler({ // Only an UNMARKED escape — today's Oura / Polar write-path throw, // or any future provider path — is recorded here. if (!isSyncFailureRecorded(err)) { - await recordFailure(userId, err).catch((recErr) => { - evt.addWarning( - `${taskName} recordFailure failed for user ${userId}: ${recErr}`, - ); + await recordFailure(userId, err).catch(() => { + evt.addWarning(`${taskName} recordFailure failed`); }); } usersFailed++; - evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); + evt.addWarning(`${taskName} failed for one cohort member`); } } diff --git a/src/lib/jobs/reminder/withings-sync.ts b/src/lib/jobs/reminder/withings-sync.ts index efcf86750..4d36d02bb 100644 --- a/src/lib/jobs/reminder/withings-sync.ts +++ b/src/lib/jobs/reminder/withings-sync.ts @@ -7,8 +7,9 @@ import { type Job } from "pg-boss"; import { fireAndForget } from "@/lib/logging/fire-and-forget"; import { recordError, recordWithingsSync } from "@/lib/jobs/worker-status"; -import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { type JobOutcome } from "@/lib/jobs/job-outcome"; import { withBackgroundEvent } from "@/lib/logging/background"; +import { isReauthRequired } from "@/lib/integrations/status"; import { retryDueWithingsWebhookSubscriptions, syncUserMeasurements, @@ -18,6 +19,10 @@ import { syncUserSleep } from "@/lib/withings/sync-sleep"; import { syncUserEcg } from "@/lib/withings/sync-ecg"; import type { WithingsEcgSyncPayload } from "@/lib/jobs/withings-ecg-queue"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; +import { + foldIntegrationCohortOutcomes, + type IntegrationUserVerdict, +} from "./poll-cohort"; import { getWorkerPrisma } from "./shared"; export type { WithingsEcgSyncPayload } from "@/lib/jobs/withings-ecg-queue"; @@ -74,52 +79,61 @@ export async function handleWithingsFallbackSync( }); if (connections.length === 0) { - return jobDone({ - total: 0, - users_synced: 0, - users_failed: 0, - measurements_imported: 0, - subscription_repair_failed: subscriptionRepairFailed, + return foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts: [], + downstreamFailures: subscriptionRepairFailed ? 1 : 0, }); } - let usersSynced = 0; - let usersFailed = 0; - let measurementsImported = 0; + const verdicts: IntegrationUserVerdict[] = []; for (const connection of connections) { + if (await isReauthRequired(connection.userId, "withings")) { + verdicts.push({ status: "parked", imported: 0 }); + continue; + } + try { - const { imported } = await syncUserMeasurements(connection.userId); - usersSynced++; - measurementsImported += imported; + const result = await syncUserMeasurements(connection.userId); + const parked = + result.failed && + result.imported === 0 && + (await isReauthRequired(connection.userId, "withings")); + verdicts.push({ + status: parked + ? "parked" + : result.failed + ? result.imported > 0 + ? "partial" + : "failed" + : "complete", + imported: result.imported, + retryable: result.failed && !parked, + }); // v1.18.1 — a fresh reading landed; resolve this user's Vorsorge // reminders eventfully. Fire-and-forget; the cron is the net. - if (imported > 0) { + if (result.imported > 0) { fireAndForget(enqueueReminderSatisfy(connection.userId), { action: "reminder.satisfy.enqueue", }); } } catch { - usersFailed++; - evt.addWarning(`Fallback sync failed for user ${connection.userId}`); + verdicts.push({ status: "failed", imported: 0, retryable: true }); + evt.addWarning("Fallback sync failed for one cohort member"); } } + const outcome = foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts, + downstreamFailures: subscriptionRepairFailed ? 1 : 0, + }); evt.setBackground({ task_name: "job.withings_sync", - result: { - users_synced: usersSynced, - total: connections.length, - measurements_imported: measurementsImported, - }, - }); - return jobDone({ - total: connections.length, - users_synced: usersSynced, - users_failed: usersFailed, - measurements_imported: measurementsImported, - subscription_repair_failed: subscriptionRepairFailed, + result: outcome.did, }); + return outcome; } catch (err) { evt.setError(err); recordError(); @@ -174,45 +188,46 @@ export async function handleWithingsEcgSync( targets.push(...connections); } if (targets.length === 0) { - return jobDone({ - total: 0, - users_synced: 0, - users_failed: 0, - recordings_imported: 0, + return foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts: [], }); } - let usersSynced = 0; - let usersFailed = 0; - let recordingsImported = 0; + const verdicts: IntegrationUserVerdict[] = []; for (const { userId, startdate, enddate } of targets) { + if (await isReauthRequired(userId, "withings")) { + verdicts.push({ status: "parked", imported: 0 }); + continue; + } + try { const options = startdate === undefined && enddate === undefined ? {} : { startdate, enddate }; - recordingsImported += await syncUserEcg(userId, options); - usersSynced++; - } catch (err) { - usersFailed++; - evt.addWarning(`Withings ECG sync failed for user ${userId}: ${err}`); + const imported = await syncUserEcg(userId, options); + const parked = + imported === 0 && (await isReauthRequired(userId, "withings")); + verdicts.push({ + status: parked ? "parked" : "complete", + imported, + }); + } catch { + verdicts.push({ status: "failed", imported: 0, retryable: true }); + evt.addWarning("Withings ECG sync failed for one cohort member"); } } + const outcome = foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts, + }); evt.setBackground({ task_name: "job.withings_ecg_sync", - result: { - users_synced: usersSynced, - total: targets.length, - recordings_imported: recordingsImported, - }, - }); - return jobDone({ - total: targets.length, - users_synced: usersSynced, - users_failed: usersFailed, - recordings_imported: recordingsImported, + result: outcome.did, }); + return outcome; } catch (err) { evt.setError(err); recordError(); @@ -257,47 +272,47 @@ export async function handleWithingsActivitySync( targets.push(...connections); } if (targets.length === 0) { - return jobDone({ - total: 0, - users_synced: 0, - users_failed: 0, - measurements_imported: 0, + return foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts: [], }); } - let usersSynced = 0; - let usersFailed = 0; - let measurementsImported = 0; + const verdicts: IntegrationUserVerdict[] = []; for (const { userId } of targets) { + if (await isReauthRequired(userId, "withings")) { + verdicts.push({ status: "parked", imported: 0 }); + continue; + } + try { const imported = await syncUserActivity(userId); - usersSynced++; - measurementsImported += imported; + const parked = + imported === 0 && (await isReauthRequired(userId, "withings")); + verdicts.push({ + status: parked ? "parked" : "complete", + imported, + }); if (imported > 0) { fireAndForget(enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } } catch { - usersFailed++; - evt.addWarning(`Withings activity sync failed for user ${userId}`); + verdicts.push({ status: "failed", imported: 0, retryable: true }); + evt.addWarning("Withings activity sync failed for one cohort member"); } } + const outcome = foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts, + }); evt.setBackground({ task_name: "job.withings_activity_sync", - result: { - users_synced: usersSynced, - total: targets.length, - measurements_imported: measurementsImported, - }, - }); - return jobDone({ - total: targets.length, - users_synced: usersSynced, - users_failed: usersFailed, - measurements_imported: measurementsImported, + result: outcome.did, }); + return outcome; } catch (err) { evt.setError(err); recordError(); @@ -330,47 +345,47 @@ export async function handleWithingsSleepSync( targets.push(...connections); } if (targets.length === 0) { - return jobDone({ - total: 0, - users_synced: 0, - users_failed: 0, - measurements_imported: 0, + return foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts: [], }); } - let usersSynced = 0; - let usersFailed = 0; - let measurementsImported = 0; + const verdicts: IntegrationUserVerdict[] = []; for (const { userId } of targets) { + if (await isReauthRequired(userId, "withings")) { + verdicts.push({ status: "parked", imported: 0 }); + continue; + } + try { const imported = await syncUserSleep(userId); - usersSynced++; - measurementsImported += imported; + const parked = + imported === 0 && (await isReauthRequired(userId, "withings")); + verdicts.push({ + status: parked ? "parked" : "complete", + imported, + }); if (imported > 0) { fireAndForget(enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } } catch { - usersFailed++; - evt.addWarning(`Withings sleep sync failed for user ${userId}`); + verdicts.push({ status: "failed", imported: 0, retryable: true }); + evt.addWarning("Withings sleep sync failed for one cohort member"); } } + const outcome = foldIntegrationCohortOutcomes({ + provider: "withings", + verdicts, + }); evt.setBackground({ task_name: "job.withings_sleep_sync", - result: { - users_synced: usersSynced, - total: targets.length, - measurements_imported: measurementsImported, - }, - }); - return jobDone({ - total: targets.length, - users_synced: usersSynced, - users_failed: usersFailed, - measurements_imported: measurementsImported, + result: outcome.did, }); + return outcome; } catch (err) { evt.setError(err); recordError();