855244747c
Two related fixes from the architecture review: Typed job seam. Job payloads were `unknown` end-to-end: every handler opened with `job.data as SomePayload`, every enqueue site passed a bare string queue name and an unchecked object, and a typo meant a runtime failure in a background worker. packages/jobs gains defineJob(), which keeps the payload typed inside the handler and performs the contravariance cast once inside the package. The journal declares all 14 queues and their payload shapes in app/jobs/payloads.ts; enqueue()/ enqueueOptional() and defineJournalJob() key off that map, so enqueue sites and handlers cannot drift and queue names are compile-checked. Komoot credential bypass. The bulk-import route enqueued the raw credentials JSONB in the pg-boss payload, skipping withFreshCredentials entirely: credentials sat at rest in the job table, were never refreshed if stale, and markNeedsRelink never fired. The payload now carries only the serviceId; the handler resolves fresh credentials through the ConnectedServiceManager at execution time, and marks the import batch failed (instead of leaving it pending forever) when credential resolution itself fails. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
85 lines
3 KiB
TypeScript
85 lines
3 KiB
TypeScript
// Process-level pg-boss singleton. server.ts initializes the boss + starts
|
|
// the worker; feature code (e.g., activities.server.ts) calls `getBoss()`
|
|
// to enqueue jobs against the same instance. The singleton's lifecycle
|
|
// is bound to the Node process — startWorker calls boss.start(); the
|
|
// SIGTERM handler stops it.
|
|
//
|
|
// The instance lives on globalThis, NOT in a module-local variable. In
|
|
// production there are TWO copies of this module in the process:
|
|
// server.ts imports the TypeScript source directly (node
|
|
// --experimental-strip-types), while route handlers live in the
|
|
// bundled build/server/index.js with its own module instance. A
|
|
// module-local `_boss` set by server.ts is invisible to the bundle, so
|
|
// every request-path enqueue (notifications fan-out, federation
|
|
// deliveries, komoot imports) silently failed in production with
|
|
// "pg-boss not initialized". Symbol.for() gives both copies the same
|
|
// global registry key.
|
|
|
|
import { logger } from "./logger.server.ts";
|
|
import type { JobName, JobPayloads } from "../jobs/payloads.ts";
|
|
|
|
// Structurally typed (we only need `send`) so we don't have to pull
|
|
// pg-boss into the journal app's dep graph just for the typedef.
|
|
export interface BossSendOptions {
|
|
retryLimit?: number;
|
|
retryBackoff?: boolean;
|
|
retryDelay?: number;
|
|
}
|
|
|
|
interface BossLike {
|
|
send(queueName: string, data: unknown, options?: BossSendOptions): Promise<string | null>;
|
|
}
|
|
|
|
const BOSS_KEY = Symbol.for("trails-cool.journal.pg-boss");
|
|
|
|
type BossGlobal = { [BOSS_KEY]?: BossLike | null };
|
|
|
|
/** Set by server.ts once the boss is created + started. */
|
|
export function setBoss(boss: BossLike | null): void {
|
|
(globalThis as BossGlobal)[BOSS_KEY] = boss;
|
|
}
|
|
|
|
/**
|
|
* Get the started pg-boss instance. Throws if called before
|
|
* server.ts has initialized it (i.e., outside a running Journal
|
|
* server context).
|
|
*/
|
|
export function getBoss(): BossLike {
|
|
const boss = (globalThis as BossGlobal)[BOSS_KEY];
|
|
if (!boss) {
|
|
throw new Error("pg-boss not initialized — getBoss called before server bootstrap");
|
|
}
|
|
return boss;
|
|
}
|
|
|
|
/**
|
|
* Enqueue a job. The queue name must be a key of JobPayloads and the
|
|
* payload must match its declared shape. Throws if the queue is down —
|
|
* use this when the caller's correctness depends on the job existing
|
|
* (e.g. a batch row that would otherwise wait forever).
|
|
*/
|
|
export async function enqueue<K extends JobName>(
|
|
queue: K,
|
|
data: JobPayloads[K],
|
|
options?: BossSendOptions,
|
|
): Promise<void> {
|
|
await getBoss().send(queue, data, options);
|
|
}
|
|
|
|
/**
|
|
* Best-effort enqueue: log + swallow errors so a downstream queue
|
|
* outage doesn't fail the user-visible request that triggered the
|
|
* fan-out. Use this for "fire and forget" notifications work.
|
|
*/
|
|
export async function enqueueOptional<K extends JobName>(
|
|
queue: K,
|
|
data: JobPayloads[K],
|
|
ctx: Record<string, unknown> = {},
|
|
options?: BossSendOptions,
|
|
): Promise<void> {
|
|
try {
|
|
await enqueue(queue, data, options);
|
|
} catch (err) {
|
|
logger.warn({ err, queue, ...ctx }, "boss.send failed; continuing");
|
|
}
|
|
}
|