diff --git a/src/lib/jobs/__tests__/job-outcome-source-inventory.test.ts b/src/lib/jobs/__tests__/job-outcome-source-inventory.test.ts new file mode 100644 index 000000000..02197386b --- /dev/null +++ b/src/lib/jobs/__tests__/job-outcome-source-inventory.test.ts @@ -0,0 +1,140 @@ +/** + * Production success-fact inventory. + * + * `runJob` now serializes every successful handler result before pg-boss + * completes the row. A newly introduced fact key that is not deliberately + * reviewed would therefore fail the job at runtime. This source-level guard + * keeps the fixed vocabulary and every production `jobDone({...})` call in + * lockstep while the serializer's negative tests keep credentials, URLs, + * identifiers, free text, nested payloads, and health values outside it. + */ +import { readdirSync, readFileSync } from "node:fs"; +import { join } from "node:path"; +import ts from "typescript"; +import { describe, expect, it } from "vitest"; + +import { JOB_FACT_ALLOWLIST } from "@/lib/jobs/job-outcome"; + +const JOBS_ROOT = join(process.cwd(), "src/lib/jobs"); + +function productionJobFiles(directory = JOBS_ROOT): string[] { + return readdirSync(directory, { withFileTypes: true }).flatMap((entry) => { + const path = join(directory, entry.name); + if (entry.isDirectory()) { + return entry.name === "__tests__" ? [] : productionJobFiles(path); + } + return entry.name.endsWith(".ts") ? [path] : []; + }); +} + +function objectFactKeys( + object: ts.ObjectLiteralExpression, + source: ts.SourceFile, +): string[] { + return object.properties.map((property) => { + if (ts.isSpreadAssignment(property) || !property.name) { + throw new Error( + `jobDone facts must have explicit keys at ${source.fileName}:${ + source.getLineAndCharacterOfPosition(property.getStart()).line + 1 + }`, + ); + } + return property.name.getText(source).replace(/^["']|["']$/g, ""); + }); +} + +function declaredObject( + source: ts.SourceFile, + name: string, +): ts.ObjectLiteralExpression | null { + let found: ts.ObjectLiteralExpression | null = null; + const visit = (node: ts.Node): void => { + if ( + ts.isVariableDeclaration(node) && + ts.isIdentifier(node.name) && + node.name.text === name && + node.initializer && + ts.isObjectLiteralExpression(node.initializer) + ) { + found = node.initializer; + } + ts.forEachChild(node, visit); + }; + visit(source); + return found; +} + +function successFactKeys(file: string): string[] { + const source = ts.createSourceFile( + file, + readFileSync(file, "utf8"), + ts.ScriptTarget.Latest, + true, + ); + const keys: string[] = []; + + const visit = (node: ts.Node): void => { + if ( + ts.isCallExpression(node) && + ts.isIdentifier(node.expression) && + node.expression.text === "jobDone" && + node.arguments.length > 0 + ) { + const facts = node.arguments[0]!; + if (ts.isObjectLiteralExpression(facts)) { + keys.push(...objectFactKeys(facts, source)); + } else if (ts.isIdentifier(facts)) { + const object = declaredObject(source, facts.text); + if (!object) { + throw new Error( + `jobDone facts must resolve to a local object at ${file}:${ + source.getLineAndCharacterOfPosition(facts.getStart()).line + 1 + }`, + ); + } + keys.push(...objectFactKeys(object, source)); + } else { + throw new Error( + `jobDone facts must be an auditable object at ${file}:${ + source.getLineAndCharacterOfPosition(facts.getStart()).line + 1 + }`, + ); + } + } + ts.forEachChild(node, visit); + }; + visit(source); + return keys; +} + +describe("durable production job-outcome fact inventory", () => { + it("allows every explicit production success-fact key", () => { + const unknown = productionJobFiles() + .flatMap((file) => + successFactKeys(file).map((key) => ({ + file: file.replace(`${process.cwd()}/`, ""), + key, + })), + ) + .filter(({ key }) => !JOB_FACT_ALLOWLIST.has(key)); + + expect(unknown).toEqual([]); + }); + + it("keeps location, credential, capability, and health identifiers out", () => { + for (const forbidden of [ + "rotate_active_key_id", + "offhost_backup_endpoint", + "offhost_backup_bucket", + "restore_drill_object_key", + "tls_pin_monitor_host", + "intake_auto_skip_cutoff", + "user_id", + "access_token", + "refresh_token", + "health_value", + ]) { + expect(JOB_FACT_ALLOWLIST.has(forbidden), forbidden).toBe(false); + } + }); +}); diff --git a/src/lib/jobs/encryption-key-rotate.ts b/src/lib/jobs/encryption-key-rotate.ts index a33588426..e8ebefe5b 100644 --- a/src/lib/jobs/encryption-key-rotate.ts +++ b/src/lib/jobs/encryption-key-rotate.ts @@ -83,7 +83,6 @@ export async function handleEncryptionKeyRotate( // deployment no longer configures. The pass still ran, so they ride out // as a count rather than failing (and retrying) the whole corpus. return jobDone({ - rotate_active_key_id: result.activeKeyId, rotate_scanned: result.totalScanned, rotate_rotated: result.totalRotated, rotate_errors: result.totalErrors, diff --git a/src/lib/jobs/environment-fetch.ts b/src/lib/jobs/environment-fetch.ts index 84c8dd7b2..d4c9cdf9c 100644 --- a/src/lib/jobs/environment-fetch.ts +++ b/src/lib/jobs/environment-fetch.ts @@ -169,7 +169,7 @@ export async function handleEnvironmentFetch( "info", `[environment-fetch] user=${userId} module disabled — skipping`, ); - return jobDone({ skipped: "environment module disabled" }); + return jobDone({ skipped: "environment_module_disabled" }); } const { startDate, endDate } = resolveRange(payload); diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 34db35a3e..2c84daf4b 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -42,32 +42,173 @@ export type JobFacts = Readonly>; * free-form error, or health value to a durable job result. */ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ - "provider", + "access_tokens_deleted", + "assessments_warmed", + "auto_resolved", + "backed", + "buckets_collapsed", + "buckets_recomputed", + "budget_blocked", + "cached", + "candidates_scanned", + "connect_tickets_deleted", + "connections_deleted", + "considered", + "created", + "daily_rows_retired", + "days_consolidated", + "days_rebuilt", + "days_recomputed", + "days_skipped", + "days_skipped_no_tombstones", + "days_stored", + "deleted", + "dense_days_consolidated", + "dense_rows_soft_deleted", + "dense_skipped", + "derived_resting_rows_upserted", + "discovery_enqueued", + "discovery_failed", + "discovery_skipped", + "dispatched", + "dispatched_period_confirm", + "dispatched_period_soon", + "document_purge_deleted", + "documents", + "documents_indexed", + "dose_changes_migrated", + "downstream_failed", + "drained", + "duration_ms", + "errored", + "failed", + "failures", + "feedback_buckets", + "feedback_total_rows", + "feedback_window_days", + "fetches", + "finalised", + "finalized", + "forced", + "generated", + "geo_backfill_carrier_resolved", + "geo_backfill_located", + "geo_backfill_scanned", + "geo_backfill_skipped", + "geo_backfill_still_unresolved", + "host_metric_pruned", + "hourly_rows_upserted", + "imported", + "in_slot", + "in_window", + "insufficient", + "intake_auto_skip_count", + "intakes_pruned", + "inventory_expired_count", + "inventory_items_migrated", + "jobs", + "legacy_rows_soft_deleted", + "linked", + "malformed", + "manual_mints_removed", + "markers", + "mean_days_consolidated", + "mean_rows_soft_deleted", + "measurements_imported", + "measurements_migrated", + "measurements_pruned", + "medications", + "medications_evaluated", + "metric_assessments_warmed", + "missed_doses_minted", + "mood_entries_migrated", + "mood_pruned", + "notifications_dispatched", + "notifications_failed", + "notified", + "offhost_backup_configured", + "offhost_backup_failed", + "offhost_backup_total_users", + "offhost_backup_uploaded", "outcome", + "passes_failed", + "per_sample_rows_deleted", + "per_sample_rows_soft_deleted", + "persisted", + "plan_reviews_minted", + "pr_detection_inserted", + "pr_detection_jobs", + "pr_detection_ties", + "pr_detection_users", + "pr_detection_users_failed", + "processed", + "provider", + "queued", + "rearmed", + "records_read", + "refreshed", + "reminders_due", + "removed", + "restore_drill_age_days", + "restore_drill_ciphertext_bytes", + "restore_drill_intake_events", + "restore_drill_measurements", + "restore_drill_medications", + "restore_drill_mood_entries", + "restore_drill_plaintext_bytes", + "restore_drill_stale", + "retried", + "reviewed", + "rollup_failed", + "rotate_errors", + "rotate_rotated", + "rotate_scanned", + "rows_deleted", + "rows_normalised", + "rows_reinserted", + "rows_resurrected", + "rows_soft_deleted", + "rows_upserted", + "satisfied", + "scheduled", + "sent", + "served", + "side_effects_migrated", + "single_user", + "skipped", + "skipped_already_dispatched", + "skipped_already_logged", + "skipped_already_notified", + "skipped_incomplete", + "skipped_no_channel", + "skipped_not_due", + "skipped_outside_window", + "slots_collapsed", + "states_deleted", + "stored", + "subscription_repair_failed", + "summaries_enqueued", + "suppressed_client_managed", + "suppressed_discreet", + "thumbnails_enqueued", + "tls_pin_monitor_known_count", + "tls_pin_monitor_outcome", "total", - "users_synced", + "unchanged", + "updated", + "users", + "users_clean_zero", "users_complete", - "users_partial", + "users_enqueued", "users_failed", "users_parked", - "users_skipped", - "users_useful", - "users_clean_zero", + "users_partial", "users_retryable", - "downstream_failed", - "measurements_imported", - "deleted", - "generated", - "jobs", - "skipped", - "processed", - "created", - "updated", - "failed", - "retried", - "queued", - "sent", - "subscription_repair_failed", + "users_scanned", + "users_skipped", + "users_synced", + "users_useful", + "wedge_skipped", ]); export const MAX_JOB_FACTS = 32; diff --git a/src/lib/jobs/reminder/backup-handlers.ts b/src/lib/jobs/reminder/backup-handlers.ts index 6d23f8a44..48d0e3e47 100644 --- a/src/lib/jobs/reminder/backup-handlers.ts +++ b/src/lib/jobs/reminder/backup-handlers.ts @@ -52,8 +52,6 @@ export async function handleOffhostBackup( offhost_backup_uploaded: report.uploaded, offhost_backup_failed: report.failed, offhost_backup_total_users: report.totalUsers, - offhost_backup_endpoint: report.config.endpoint, - offhost_backup_bucket: report.config.bucket, }); } catch (err) { // Not configured ⇒ skip silently with a warning, not an error: most diff --git a/src/lib/jobs/reminder/medication-maintenance.ts b/src/lib/jobs/reminder/medication-maintenance.ts index b906eb8f6..14a6a12ea 100644 --- a/src/lib/jobs/reminder/medication-maintenance.ts +++ b/src/lib/jobs/reminder/medication-maintenance.ts @@ -64,7 +64,6 @@ export async function handleIntakeAutoSkip( evt.addMeta("intake_auto_skip_cutoff", result.cutoff.toISOString()); return jobDone({ intake_auto_skip_count: result.skippedCount, - intake_auto_skip_cutoff: result.cutoff.toISOString(), }); } catch (err) { evt.addWarning( diff --git a/src/lib/jobs/reminder/ops-handlers.ts b/src/lib/jobs/reminder/ops-handlers.ts index b483243c5..a8774641e 100644 --- a/src/lib/jobs/reminder/ops-handlers.ts +++ b/src/lib/jobs/reminder/ops-handlers.ts @@ -149,11 +149,11 @@ export async function handleTlsPinMonitor( evt.addMeta("tls_pin_monitor_outcome", summary.outcome); evt.addMeta("tls_pin_monitor_host", summary.host); evt.addMeta("tls_pin_monitor_known_count", summary.knownPinCount); - // No configured target leaves `host` null; the empty string keeps the - // fact shape scalar, and `outcome: "skipped"` already names that case. + // The configured host remains in the scoped operator event above. It is + // deliberately excluded from the durable pg-boss result, which carries + // only the stable outcome code and non-identifying pin count. return jobDone({ tls_pin_monitor_outcome: summary.outcome, - tls_pin_monitor_host: summary.host ?? "", tls_pin_monitor_known_count: summary.knownPinCount, }); } catch (err) { diff --git a/src/lib/jobs/restore-drill.ts b/src/lib/jobs/restore-drill.ts index 5497174ce..36056c56f 100644 --- a/src/lib/jobs/restore-drill.ts +++ b/src/lib/jobs/restore-drill.ts @@ -189,7 +189,6 @@ export async function handleRestoreDrill( // above already reached the operator. Failing the job here would retry a // reading that cannot change until the next upload lands. return jobDone({ - restore_drill_object_key: report.objectKey, restore_drill_age_days: report.ageDays, restore_drill_ciphertext_bytes: report.ciphertextBytes, restore_drill_plaintext_bytes: report.plaintextBytes, @@ -204,7 +203,7 @@ export async function handleRestoreDrill( evt.addWarning(`restore-drill skipped: ${err.message}`); // Most self-hosters never set the S3 vars. Nothing to verify is not a // failed verification, and a monthly failed job would be noise. - return jobDone({ skipped: "offhost backup not configured" }); + return jobDone({ skipped: "offhost_backup_not_configured" }); } evt.addWarning(`restore-drill failed: ${err}`); await reportWorkerError(RESTORE_DRILL_QUEUE, err); diff --git a/tests/integration/job-success-outcome.test.ts b/tests/integration/job-success-outcome.test.ts index 08255a9af..bfa5c3ecd 100644 --- a/tests/integration/job-success-outcome.test.ts +++ b/tests/integration/job-success-outcome.test.ts @@ -6,7 +6,7 @@ * test deliberately registers the production wrapper as the real worker * callback and reads the completed row back through pg-boss. */ -import { PgBoss, type Job } from "pg-boss"; +import { PgBoss } from "pg-boss"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { jobDone } from "@/lib/jobs/job-outcome"; @@ -51,7 +51,7 @@ describe("runJob durable pg-boss success output", () => { await boss.work>( QUEUE, { pollingIntervalSeconds: 0.5 }, - runJob(QUEUE, async (_jobs: Job[]) => + runJob(QUEUE, async () => jobDone({ provider: "google_health", outcome: "useful",