/**
 * @factory/observability — OTel SDK init (TASK-0135).
 *
 * `initOtel()` is the single entry the four apps call at boot. Contract:
 *
 *  - DISABLED BY DEFAULT: when `OTEL_EXPORTER_OTLP_ENDPOINT` is unset (or
 *    empty) it is a NO-OP — dev/tests run with zero collector dependency
 *    (the NFR-H5 self-host posture: telemetry is opt-in, no data leaves the
 *    deployment). When disabled, `isOtelEnabled()` is false and every
 *    app-side helper degrades to its no-op path.
 *  - IDEMPOTENT: a second call (Nest re-init under vitest, HMR) returns the
 *    first state; it never double-registers the global providers.
 *  - Traces: BatchSpanProcessor (wrapped by the span-scrubber) → OTLP/HTTP;
 *    sampling = ParentBased over a 10% TraceIdRatio root sampler (NFR §7.1
 *    "100% errors, 10% others" — the 10% head-sample is the v1 shape; the
 *    collector-side error-always tail sampling is out of scope, recorded
 *    decision).
 *  - Metrics: PeriodicExportingMetricReader (30 s) → OTLP/HTTP; the SDK's
 *    MeterProvider is registered GLOBALLY so the singleton `metrics`
 *    registry (metrics.ts) exports to it.
 *  - Propagation: the package's W3C trace-context propagator (traceparent.ts)
 *    is registered so `captureTraceparent()`/`resumeTraceparent()` behave
 *    identically before AND after the SDK boots (the API→worker job-metadata
 *    primitive).
 *  - Context: AsyncLocalStorageContextManager (required for the
 *    API→worker traceparent primitive to work across the async queue
 *    boundary + for auto-instrumented spans to parent correctly).
 *  - Instruments: `HttpInstrumentation` (outgoing calls to OpenSearch/MinIO
 *    auto-continue the trace — the spec's "API → OpenSearch/MinIO (HTTP
 *    propagation)") + the redis instrumentations (spans for ioredis /
 *    node-redis commands). Per-Redis-command DURATIONS are recorded
 *    explicitly from the apps via the `metrics` registry regardless —
 *    deterministic, no require-order dependence (recorded decision).
 *  - Logs: v1 keeps pino → stdout (NFR §7.1: "pino → stdout (JSON)" is
 *    primary; Loki scrape is the staging convenience path). The `logs`
 *    option is accepted for spec-completeness and NOT wired (recorded gap).
 *
 * The SDK is imported lazily (only when enabled) so a disabled process
 * never pays the sdk import cost and never touches the global API.
 *
 * OTel JS 2.x surface (T-190b — the versions pinned in package.json: SDK
 * core 2.12.0 + exporters 0.223.0 + `@prisma/instrumentation` 6.19.3, which
 * pairs with the 2.x train and clears the two audit exceptions carried under
 * T-190): `NodeTracerProvider` takes its span processors via the
 * CONSTRUCTOR `spanProcessors` option (the 1.x `addSpanProcessor()` pair is
 * gone), resources are built with `resourceFromAttributes(...)` (the
 * `Resource` class and `Resource.default().merge(...)` are gone),
 * `SpanStatusCode` still comes from the API, and the OTLP/HTTP exporters
 * still take a `url` option.
 *
 * SUPERSEDED (1.x pairing, recorded 2026-10): the SDK previously pinned the
 * 1.30.x line because `@prisma/instrumentation@5.22.0` (Prisma 5.22) built
 * its engine spans with a 1.x `new Span(...)` that crash-looped against a
 * 2.x global tracer (`parentTracer.getSpanLimits is not a function`). That
 * constraint dissolved when Prisma moved to 6.x — `@prisma/instrumentation`
 * 6.x is built against the 2.x SDK, so the whole train moves to
 * 2.12.0/0.223.0 (T-190b). Per AGENTS rule 29 the superseded decision is
 * kept in git history and the FEATURE_TRACKER T-190b entry, not in this
 * file's live rationale.
 */
import {
  context as otelContext,
  metrics as metricsApi,
  propagation,
  SpanStatusCode,
  trace,
  type Context,
  type Span
} from "@opentelemetry/api";
import type { MeterProvider as SdkMeterProvider } from "@opentelemetry/sdk-metrics";
import { BasicTracerProvider, type ReadableSpan, type Span as SdkSpan, type SpanProcessor } from "@opentelemetry/sdk-trace-base";
import { buildResource, type ServiceName } from "./resource.js";
import { registerTraceparentPropagator, traceparentPropagator } from "./traceparent.js";
import { spanScrubber } from "./scrubber.js";
import { m } from "./metrics.js";
import { setPrometheusReader } from "./prometheus.js";

export interface InitOtelOptions {
  /** One of the four factory service names (the spec's list). */
  serviceName: ServiceName;
  /** OTLP/HTTP endpoint (e.g. `http://otel-collector:4318`). Unset/empty → disabled. */
  endpoint?: string;
  /** See module docs: accepted, not wired in v1 (pino → stdout is primary). */
  logs?: boolean;
  /** Root-span sample ratio for NEW traces (0..1). Default 0.1 (NFR §7.1). */
  sampleRatio?: number;
  /** Metric export interval (ms). Default 30_000. */
  metricExportIntervalMs?: number;
  /** `deployment.environment` attribute (defaults to NODE_ENV). */
  environment?: string;
}

export interface OtelState {
  /** False when the endpoint was unset — every consumer must no-op. */
  enabled: boolean;
  serviceName: ServiceName;
  tracerProvider?: BasicTracerProvider;
  meterProvider?: SdkMeterProvider;
}

let state: OtelState | undefined;

/** Whether telemetry is live (the endpoint was set at init time). */
export function isOtelEnabled(): boolean {
  return state?.enabled === true;
}

/** The current init state (undefined = never initialized → disabled). */
export function getOtelState(): OtelState | undefined {
  return state;
}

/**
 * SpanProcessor wrapper that scrubs PII + secrets off every span's
 * attributes at `onEnd` — BEFORE the BatchSpanProcessor queues it for
 * export. This is the central scrubbing point: it covers spans this package
 * creates AND the auto-instrumented ones (HTTP, ioredis, Prisma — the
 * Prisma span carries `db.*` attributes the scrubber knows to redact).
 *
 * `onStart` receives the SDK `Span`, `onEnd` receives the `ReadableSpan`;
 * both are handled via the shared inner processor.
 */
class ScrubbingSpanProcessor implements SpanProcessor {
  constructor(private readonly inner: SpanProcessor) {}

  forceFlush(): Promise<void> {
    return this.inner.forceFlush();
  }

  shutdown(): Promise<void> {
    return this.inner.shutdown();
  }

  onStart(span: SdkSpan, parentContext: Context): void {
    this.inner.onStart(span, parentContext);
  }

  onEnd(span: ReadableSpan): void {
    if (span.attributes) spanScrubber(span.attributes as unknown as Record<string, unknown>);
    this.inner.onEnd(span);
  }
}

/**
 * Central db/redis duration collector (TASK-0135). Records, on `onEnd`, the
 * two spec histograms that would otherwise need an edit at EVERY Prisma /
 * ioredis call site:
 *
 *  - Prisma operation spans (from `@prisma/instrumentation`, active only when
 *    the schema has `previewFeatures = ["tracing"]`): span name
 *    `prisma:client:operation`, attributes `model` + `method` (= the Prisma
 *    action) → `db_query_duration_seconds{model, operation}`.
 *  - ioredis command spans: span name = the command, attribute
 *    `db.system.name = "redis"` → `redis_command_duration_seconds{command}`.
 *
 * Duration comes from `span.duration` (HrTime `[sec, nanos]`) — no
 * manual start/end math. The instruments are the singleton registry's
 * (metrics.ts) — no-op when OTel is disabled, so a span ending can never
 * throw here. Order in the chain: AFTER the scrubber (attributes are already
 * redacted; we only READ model/operation/command, never the query text).
 */
export class DbRedisMetricsProcessor implements SpanProcessor {
  forceFlush(): Promise<void> {
    return Promise.resolve();
  }

  shutdown(): Promise<void> {
    return Promise.resolve();
  }

  onStart(): void {
    /* metrics are recorded at end only */
  }

  onEnd(span: ReadableSpan): void {
    const seconds = hrTimeToSeconds(span.duration);
    const attrs = (span.attributes ?? {}) as Record<string, unknown>;
    if (span.name === "prisma:client:operation") {
      const model = typeof attrs["model"] === "string" ? (attrs["model"] as string) : "unknown";
      const operation = typeof attrs["method"] === "string" ? (attrs["method"] as string) : "unknown";
      m.db.record({ model, operation }, seconds);
      return;
    }
    if (attrs["db.system.name"] === "redis") {
      const command = typeof attrs["db.operation.name"] === "string" ? (attrs["db.operation.name"] as string) : span.name;
      m.redis.record({ command }, seconds);
    }
  }
}

/** HrTime `[seconds, nanos]` → fractional seconds (`ReadableSpan.duration`, same shape in 1.x). */
function hrTimeToSeconds(hrTime: readonly [number, number]): number {
  return hrTime[0] + hrTime[1] / 1_000_000_000;
}

/**
 * Initialize OTel for one app (module docs for the contract). Returns the
 * shared state; disabled when no endpoint is configured.
 */
export async function initOtel(opts: InitOtelOptions): Promise<OtelState> {
  if (state) return state; // idempotent — first init wins
  // The propagator registers even when disabled: capture/resume must behave
  // (as no-ops) identically in dev and prod.
  registerTraceparentPropagator();

  const endpoint = (opts.endpoint ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "").trim();
  if (!endpoint) {
    state = { enabled: false, serviceName: opts.serviceName };
    return state;
  }

  // Lazy import: only an ENABLED process loads the SDK.
  const [
    sdkTraceBase,
    { NodeTracerProvider },
    { OTLPTraceExporter },
    sdkMetrics,
    { OTLPMetricExporter },
    { PrometheusExporter },
    { defaultResource },
    { AsyncLocalStorageContextManager },
    httpInst,
    ioredisInst,
    redisInst,
    prismaInst
  ] = await Promise.all([
    import("@opentelemetry/sdk-trace-base"),
    import("@opentelemetry/sdk-trace-node"),
    import("@opentelemetry/exporter-trace-otlp-http"),
    import("@opentelemetry/sdk-metrics"),
    import("@opentelemetry/exporter-metrics-otlp-http"),
    import("@opentelemetry/exporter-prometheus"),
    import("@opentelemetry/resources"),
    import("@opentelemetry/context-async-hooks"),
    import("@opentelemetry/instrumentation-http"),
    import("@opentelemetry/instrumentation-ioredis"),
    import("@opentelemetry/instrumentation-redis"),
    import("@prisma/instrumentation")
  ]);

  const { BatchSpanProcessor, ParentBasedSampler, TraceIdRatioBasedSampler } = sdkTraceBase;
  const { MeterProvider, PeriodicExportingMetricReader } = sdkMetrics;

  // 2.x: resources are plain objects via resourceFromAttributes(); the
  // defaultResource() factory replaces Resource.default(). Merge our stable
  // attributes onto the SDK default (telemetry.sdk.* + detected platform
  // attrs).
  const resource = defaultResource().merge(
    buildResource(opts.serviceName, opts.environment !== undefined ? { environment: opts.environment } : {})
  );
  const sampler = new ParentBasedSampler({
    root: new TraceIdRatioBasedSampler(opts.sampleRatio ?? 0.1)
  });

  const contextManager = new AsyncLocalStorageContextManager();
  // 2.x: span processors attach via the CONSTRUCTOR `spanProcessors` option
  // (the 1.x addSpanProcessor() pair is gone). Order preserved: (1) the
  // central db/redis metric recorder, (2) the scrubber wrapping the batched
  // OTLP exporter (scrub before export).
  const tracerProvider = new NodeTracerProvider({
    resource,
    sampler,
    spanProcessors: [
      new DbRedisMetricsProcessor(),
      new ScrubbingSpanProcessor(new BatchSpanProcessor(new OTLPTraceExporter({ url: endpoint + "/v1/traces" })))
    ]
  });
  contextManager.enable();
  // Global registration (guarded above — one init per process). The
  // context manager + propagator are set explicitly (register() would also
  // set them, but the constructor already owns the processor chain).
  trace.setGlobalTracerProvider(tracerProvider);
  otelContext.setGlobalContextManager(contextManager);
  propagation.setGlobalPropagator(traceparentPropagator);

  // The Prometheus reader runs SERVER-LESS (preventServerStart) — each app
  // exposes it on its OWN port's /metrics via renderPrometheus() (the scrape
  // config targets the app ports, not a second port). The serializer is
  // constructed directly in renderPrometheus() anyway (no exporter-level
  // withoutScopeInfo/withoutTargetInfo flags used here).
  const promReader = new PrometheusExporter({ preventServerStart: true });
  setPrometheusReader(promReader);

  const meterProvider: SdkMeterProvider = new MeterProvider({
    resource,
    readers: [
      new PeriodicExportingMetricReader({
        exporter: new OTLPMetricExporter({ url: endpoint + "/v1/metrics" }),
        exportIntervalMillis: opts.metricExportIntervalMs ?? 30_000
      }),
      promReader
    ]
  });
  metricsApi.setGlobalMeterProvider(meterProvider);

  // Outgoing HTTP (OpenSearch/MinIO/AI-provider calls continue the trace).
  new httpInst.HttpInstrumentation().enable();
  // Redis spans (ioredis is the client of record; node-redis kept for the
  // redis-common shared path). Failures are non-fatal — the apps record
  // redis durations explicitly via the `metrics` registry either way.
  try {
    new ioredisInst.IORedisInstrumentation().enable();
  } catch {
    /* client not loaded yet — durations are still recorded explicitly */
  }
  try {
    new redisInst.RedisInstrumentation().enable();
  } catch {
    /* node-redis is not an app dependency — no-op */
  }
  // Prisma: enable() sets globalThis.PRISMA_INSTRUMENTATION = { helper }; the
  // generated client reads that global LAZILY at query time (verified in
  // @prisma/client 6.x), so this works even though every app's PrismaClient
  // is constructed AFTER initOtel. In 6.x the tracing helper is always built
  // (the `tracing` preview feature was promoted to GA in 6.1) — Spans →
  // DbRedisMetricsProcessor (db_query_duration_seconds) + ScrubbingSpanProcessor
  // (redacted export).
  try {
    new prismaInst.PrismaInstrumentation().enable();
  } catch {
    /* never fatal */
  }

  state = { enabled: true, serviceName: opts.serviceName, tracerProvider, meterProvider };
  return state;
}

/** Gracefully shut the SDK down (flush pending spans + metrics). */
export async function shutdownOtel(): Promise<void> {
  if (!state?.enabled) return;
  try {
    await state.tracerProvider?.shutdown();
    await state.meterProvider?.shutdown();
  } finally {
    state = undefined;
  }
}

function scrubAttrs(attributes: Record<string, unknown>): Record<string, string | number | boolean> {
  return spanScrubber({ ...attributes }) as Record<string, string | number | boolean>;
}

/**
 * Start a span with scrubbed attributes, run `fn` UNDER it (the active span
 * is set on the ambient context so anything `fn` calls — including
 * auto-instrumented spans — parents correctly), and end the span when `fn`
 * settles. Errors mark the span ERROR + rethrow. When OTel is disabled the
 * global tracer is the no-op tracer and `context.with` runs ambient — the
 * no-op contract.
 *
 * Attributes are scrubbed at CREATION (this helper) AND at `onEnd` (the
 * ScrubbingSpanProcessor) — defense in depth against attributes added by
 * auto-instrumentation mid-span.
 */
export function startSpan<T>(name: string, attributes: Record<string, unknown>, fn: (span: SpanHandle) => T): T {
  const span = trace.getTracer("factory").startSpan(name, { attributes: scrubAttrs(attributes) });
  const handle = makeSpanHandle(span);
  try {
    return otelContext.with(trace.setSpan(otelContext.active(), span), () => fn(handle));
  } catch (err) {
    markError(span, err);
    throw err;
  } finally {
    span.end();
  }
}

/**
 * Async variant of {@link startSpan} for `async` callbacks (the worker
 * processor path) — the span is ENDED when the promise settles, which the
 * sync helper cannot know.
 */
export async function startAsyncSpan<T>(
  name: string,
  attributes: Record<string, unknown>,
  fn: (span: SpanHandle) => Promise<T>
): Promise<T> {
  const span = trace.getTracer("factory").startSpan(name, { attributes: scrubAttrs(attributes) });
  const handle = makeSpanHandle(span);
  try {
    return await otelContext.with(trace.setSpan(otelContext.active(), span), () => fn(handle));
  } catch (err) {
    markError(span, err);
    throw err;
  } finally {
    span.end();
  }
}

function markError(span: Span, err: unknown): void {
  try {
    span.recordException(err instanceof Error ? err : new Error(String(err)));
    span.setStatus({ code: SpanStatusCode.ERROR, message: err instanceof Error ? err.message : String(err) });
  } catch {
    /* the no-op span never throws — keep the rethrow clean */
  }
}

function makeSpanHandle(span: Span): SpanHandle {
  return {
    attributes: {} as Record<string, unknown>,
    setAttribute: (k, v) => {
      if (v === undefined) return;
      span.setAttribute(k, v);
    },
    setStatus: (status, message) =>
      span.setStatus({
        code: status === "ok" ? SpanStatusCode.OK : SpanStatusCode.ERROR,
        ...(message !== undefined ? { message } : {})
      })
  };
}

/**
 * The handle passed to a `startSpan` callback: an attribute setter (values
 * are written straight onto the live span; the onEnd scrub re-runs over the
 * final map) + a status setter. (The raw `Span` is intentionally NOT
 * exposed — call sites that need `span.end()` timing should use
 * `startAsyncSpan`.)
 */
export interface SpanHandle {
  /**
   * The span's attribute map as of creation (already scrubbed). The live
   * span is the source of truth for anything set via `setAttribute`.
   */
  attributes: Record<string, unknown>;
  setAttribute(key: string, value: string | number | boolean): void;
  setStatus(status: "ok" | "error", message?: string): void;
}