1/** 2 * apps/worker — per-job OTel observability (TASK-0135). 3 * 4 * Wraps a BullMQ processor so EVERY job gets, with one central helper (AGENTS 5 * rule 5 — the seven processor factories must not each re-derive this): 6 * 7 * 1. The API request's trace RESUMED: the API stores the active W3C 8 * traceparent on the job's `telemetry.metadata` carrier (BullMQ 5's 9 * context-propagation field, persisted as the `tm` Redis field; see 10 * apps/api/src/common/trace-metadata.ts). We resume it FIRST so (2) is a 11 * child of the enqueueing request's span — one trace across 12 * api → queue → worker. No carrier (OTel off, background enqueue, unsampled) 13 * → the span simply has no parent (the no-op contract). 14 * 2. A `job.process` SPAN (the spec's "span per BullMQ job") with `queue` / 15 * `jobId` / `name` + `workspaceId` when the job data carries one (ABSENT — 16 * not "unknown" — when it does not; the span-attribute contract). 17 * 3. The `queue_job_duration_seconds{queue}` HISTOGRAM (recorded on both 18 * success and failure — a failed attempt still consumed queue time; the 19 * span carries the error). 20 * 21 * The wrapper rethrows on failure: BullMQ's retry policy (attempts/backoff in 22 * queues.ts) is untouched — observability must never change job semantics. 23 * When OTel is disabled every primitive here is a no-op (free calls). 24 */ 25import type { Job } from "bullmq"; 26import { m, resumeTraceparent, startAsyncSpan, telemetryTraceparent } from "@factory/observability"; 27 28export type JobProcessor = (job: Job) => Promise<unknown>; 29 30export function withJobOtel(queue: string, processor: JobProcessor): JobProcessor { 31 return async (job: Job) => { 32 const carrier = job.opts?.telemetry?.metadata; 33 const tp = telemetryTraceparent(carrier); 34 const data = job.data as Record<string, unknown> | undefined; 35 const workspaceId = data && typeof data.workspaceId === "string" ? data.workspaceId : undefined; 36 const started = Date.now(); 37 try { 38 // resume FIRST (rebuilds the parent context from the stored traceparent, 39 // so the span below is a child of the API request's span), THEN create 40 // the span under the resumed context. 41 return await resumeTraceparent( 42 tp, 43 () => 44 startAsyncSpan( 45 "job.process", 46 { queue, jobId: job.id ?? "", name: job.name, ...(workspaceId ? { workspaceId } : {}) }, 47 () => processor(job) 48 ) 49 ); 50 } finally { 51 m.queue.recordJob({ queue }, (Date.now() - started) / 1000); 52 } 53 }; 54} 55