diff --git a/src/lib/jobs/__tests__/run-job.test.ts b/src/lib/jobs/__tests__/run-job.test.ts index 6973cb89a..4558ccea9 100644 --- a/src/lib/jobs/__tests__/run-job.test.ts +++ b/src/lib/jobs/__tests__/run-job.test.ts @@ -28,21 +28,60 @@ beforeEach(() => { }); describe("runJob", () => { - it("resolves and reports nothing when the handler says it did the work", async () => { + it("returns a bounded useful result and reports nothing on success", async () => { const wrapped = runJob("rate-limit-cleanup", async () => - jobDone({ deleted: 12 }), + jobDone({ outcome: "useful", deleted: 12 }), ); - await expect(wrapped(noJobs)).resolves.toBeUndefined(); + await expect(wrapped(noJobs)).resolves.toEqual({ + ok: true, + did: { outcome: "useful", deleted: 12 }, + }); expect(reportWorkerError).not.toHaveBeenCalled(); }); - it("treats a run with nothing to do as a success", async () => { + it("returns a durable clean-zero result when there was nothing to do", async () => { const wrapped = runJob("mood-reminder-cleanup", async () => - jobDone({ deleted: 0 }), + jobDone({ outcome: "clean_zero", deleted: 0 }), ); - await expect(wrapped(noJobs)).resolves.toBeUndefined(); + await expect(wrapped(noJobs)).resolves.toEqual({ + ok: true, + did: { outcome: "clean_zero", deleted: 0 }, + }); + expect(reportWorkerError).not.toHaveBeenCalled(); + }); + + it("preserves an honest bounded partial outcome", async () => { + const wrapped = runJob("google-health-sync", async () => + jobDone({ + outcome: "partial", + users_complete: 2, + users_partial: 1, + users_failed: 1, + }), + ); + + await expect(wrapped(noJobs)).resolves.toEqual({ + ok: true, + did: { + outcome: "partial", + users_complete: 2, + users_partial: 1, + users_failed: 1, + }, + }); + expect(reportWorkerError).not.toHaveBeenCalled(); + }); + + it("refuses to persist credentials or unbounded facts on success", async () => { + const wrapped = runJob("google-health-sync", async () => + jobDone({ + refresh_token: "private-token", + } as unknown as Record), + ); + + await expect(wrapped(noJobs)).rejects.toThrow(/fact key is not allowed/i); expect(reportWorkerError).not.toHaveBeenCalled(); }); diff --git a/src/lib/jobs/run-job.ts b/src/lib/jobs/run-job.ts index 05a180084..980bc5a1d 100644 --- a/src/lib/jobs/run-job.ts +++ b/src/lib/jobs/run-job.ts @@ -20,7 +20,13 @@ */ import type { Job, JobWithMetadata } from "pg-boss"; -import type { JobFact, JobFacts, JobOutcome } from "./job-outcome"; +import { + serializeJobOutcome, + type JobFact, + type JobFacts, + type JobOutcome, + type SerializedJobOutcome, +} from "./job-outcome"; import { reportWorkerError } from "./report-worker-error"; /** A converted handler: it says what it did instead of just finishing. */ @@ -61,8 +67,9 @@ function factMeta(did: JobFacts | undefined): Record { } /** - * Wrap a handler for `boss.work`. Resolves on `ok: true`; reports and - * rethrows on `ok: false`. + * Wrap a handler for `boss.work`. Resolves with the bounded, redacted + * persistence representation on `ok: true`; reports and rethrows on + * `ok: false`. * * A handler that throws is left alone — the throw already fails the job, and * re-wrapping it would only bury the original stack. `runJob` exists for the @@ -71,10 +78,10 @@ function factMeta(did: JobFacts | undefined): Record { export function runJob( queue: string, handler: (jobs: J[]) => Promise, -): (jobs: J[]) => Promise { - return async (jobs: J[]): Promise => { +): (jobs: J[]) => Promise { + return async (jobs: J[]): Promise => { const outcome = await handler(jobs); - if (outcome.ok) return; + if (outcome.ok) return serializeJobOutcome(outcome); const failure = new JobFailure(queue, outcome.reason, outcome.cause); const causeMessage = diff --git a/tests/integration/job-success-outcome.test.ts b/tests/integration/job-success-outcome.test.ts new file mode 100644 index 000000000..08255a9af --- /dev/null +++ b/tests/integration/job-success-outcome.test.ts @@ -0,0 +1,77 @@ +/** + * Durable success-output contract against pg-boss's own schema and worker. + * + * A unit test can prove `runJob` returns a value, but only pg-boss can prove + * that the value reaches the completed job row's `output` JSONB column. This + * 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 { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import { jobDone } from "@/lib/jobs/job-outcome"; +import { runJob } from "@/lib/jobs/run-job"; + +const QUEUE = `job-success-outcome-${Date.now()}-${Math.floor( + Math.random() * 1e6, +)}`; + +let boss: PgBoss; + +async function waitForCompletedOutput( + jobId: string, +): Promise> { + const deadline = Date.now() + 15_000; + while (Date.now() < deadline) { + const job = await boss.getJobById(QUEUE, jobId); + if (job?.state === "completed") { + return (job.output ?? {}) as Record; + } + await new Promise((resolve) => setTimeout(resolve, 50)); + } + throw new Error("pg-boss job did not complete before the test deadline"); +} + +beforeAll(async () => { + boss = new PgBoss({ + connectionString: process.env.DATABASE_URL, + supervise: false, + schedule: false, + }); + await boss.start(); + await boss.createQueue(QUEUE); +}, 120_000); + +afterAll(async () => { + await boss?.stop({ graceful: false, close: true }); +}); + +describe("runJob durable pg-boss success output", () => { + it("persists only the bounded serialized outcome on completion", async () => { + await boss.work>( + QUEUE, + { pollingIntervalSeconds: 0.5 }, + runJob(QUEUE, async (_jobs: Job[]) => + jobDone({ + provider: "google_health", + outcome: "useful", + users_complete: 1, + measurements_imported: 4, + }), + ), + ); + + const jobId = await boss.send(QUEUE, {}); + if (!jobId) throw new Error("queue rejected the integration-test job"); + + await expect(waitForCompletedOutput(jobId)).resolves.toEqual({ + ok: true, + did: { + provider: "google_health", + outcome: "useful", + users_complete: 1, + measurements_imported: 4, + }, + }); + }); +});