/**
 * apps/worker — per-job OTel observability (TASK-0135).
 *
 * Wraps a BullMQ processor so EVERY job gets, with one central helper (AGENTS
 * rule 5 — the seven processor factories must not each re-derive this):
 *
 *  1. The API request's trace RESUMED: the API stores the active W3C
 *     traceparent on the job's `telemetry.metadata` carrier (BullMQ 5's
 *     context-propagation field, persisted as the `tm` Redis field; see
 *     apps/api/src/common/trace-metadata.ts). We resume it FIRST so (2) is a
 *     child of the enqueueing request's span — one trace across
 *     api → queue → worker. No carrier (OTel off, background enqueue, unsampled)
 *     → the span simply has no parent (the no-op contract).
 *  2. A `job.process` SPAN (the spec's "span per BullMQ job") with `queue` /
 *     `jobId` / `name` + `workspaceId` when the job data carries one (ABSENT —
 *     not "unknown" — when it does not; the span-attribute contract).
 *  3. The `queue_job_duration_seconds{queue}` HISTOGRAM (recorded on both
 *     success and failure — a failed attempt still consumed queue time; the
 *     span carries the error).
 *
 * The wrapper rethrows on failure: BullMQ's retry policy (attempts/backoff in
 * queues.ts) is untouched — observability must never change job semantics.
 * When OTel is disabled every primitive here is a no-op (free calls).
 */
import type { Job } from "bullmq";
import { m, resumeTraceparent, startAsyncSpan, telemetryTraceparent } from "@factory/observability";

export type JobProcessor = (job: Job) => Promise<unknown>;

export function withJobOtel(queue: string, processor: JobProcessor): JobProcessor {
  return async (job: Job) => {
    const carrier = job.opts?.telemetry?.metadata;
    const tp = telemetryTraceparent(carrier);
    const data = job.data as Record<string, unknown> | undefined;
    const workspaceId = data && typeof data.workspaceId === "string" ? data.workspaceId : undefined;
    const started = Date.now();
    try {
      // resume FIRST (rebuilds the parent context from the stored traceparent,
      // so the span below is a child of the API request's span), THEN create
      // the span under the resumed context.
      return await resumeTraceparent(
        tp,
        () =>
          startAsyncSpan(
            "job.process",
            { queue, jobId: job.id ?? "", name: job.name, ...(workspaceId ? { workspaceId } : {}) },
            () => processor(job)
          )
      );
    } finally {
      m.queue.recordJob({ queue }, (Date.now() - started) / 1000);
    }
  };
}