pike/glyph Public Public: anyone can read this; nobody can change it.

/**

 * OpenTelemetry in one file, with nothing to install: W3C trace context in

 * and out, spans, metrics and logs exported as OTLP/JSON over plain `fetch`. The

 * forge traces its own requests with it, and every app the forge scaffolds

 * carries this same file as `src/otel.ts` — one implementation, read by both.

 *

 * Configured by the standard variables, so any OpenTelemetry collector works:

 *

 *   OTEL_EXPORTER_OTLP_ENDPOINT   e.g. http://otel-collector:4318 (unset = off)

 *   OTEL_SERVICE_NAME             what this process is called in a trace

 *   OTEL_RESOURCE_ATTRIBUTES      k=v,k=v added to every span and metric

 *   OTEL_SDK_DISABLED=true        off, whatever else is set

 *

 * With no endpoint every call here is a cheap no-op: spans are still made

 * (so code reads the same) but nothing is kept or sent. Export never throws

 * and never blocks a request; a collector that is down costs a dropped

 * batch, not an error page. The queue is bounded — the oldest spans go first.

 */

import { AsyncLocalStorage } from "node:async_hooks";

import { randomBytes } from "node:crypto";

// The real fetch, taken before anyone (an app's own setup) can replace the

// global with `tracedFetch`: export and tracedFetch's own call use it, so

// neither traces itself.

const nativeFetch: typeof fetch = globalThis.fetch.bind(globalThis);

export type Attr = string | number | boolean;

export type Attrs = Record<string, Attr | undefined>;

export const SpanKind = { internal: 1, server: 2, client: 3 } as const;

export type SpanKindName = keyof typeof SpanKind;

export interface SpanContext {

  readonly traceId: string;

  readonly spanId: string;

  readonly sampled: boolean;

}

interface Config {

  readonly endpoint: string | undefined;

  readonly service: string;

  readonly resource: Attrs;

}

const env = (name: string): string | undefined => {

  const value = process.env[name]?.trim();

  return value === "" ? undefined : value;

};

/** `k=v,k=v` as the spec writes it (values may be percent-encoded). */

export function parseResourceAttributes(raw: string | undefined): Attrs {

  const out: Attrs = {};

  for (const pair of (raw ?? "").split(",")) {

    const at = pair.indexOf("=");

    if (at <= 0) continue;

    const key = pair.slice(0, at).trim();

    let value = pair.slice(at + 1).trim();

    try {

      value = decodeURIComponent(value);

    } catch {

      // A stray % is kept as written.

    }

    if (key !== "") out[key] = value;

  }

  return out;

}

function configFromEnv(): Config {

  const disabled = env("OTEL_SDK_DISABLED") === "true";

  const resource = parseResourceAttributes(env("OTEL_RESOURCE_ATTRIBUTES"));

  const service =

    env("OTEL_SERVICE_NAME") ??

    (typeof resource["service.name"] === "string"

      ? resource["service.name"]

      : "unknown_service:bun");

  return {

    endpoint: disabled

      ? undefined

      : env("OTEL_EXPORTER_OTLP_ENDPOINT")?.replace(/\/+$/, ""),

    service,

    resource,

  };

}

let config: Config = configFromEnv();

/** Re-read the environment, or set explicitly (tests, embedders). */

export function configure(next?: Partial<Config>): void {

  config = { ...configFromEnv(), ...next };

}

export const enabled = (): boolean => config.endpoint !== undefined;

// ---- trace context ----------------------------------------------------------

const hex = (bytes: number): string => randomBytes(bytes).toString("hex");

const ZERO_TRACE = "0".repeat(32);

const ZERO_SPAN = "0".repeat(16);

/** `00-<trace>-<span>-<flags>`, or undefined for anything malformed. */

export function parseTraceparent(

  header: string | null | undefined,

): SpanContext | undefined {

  const match = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/.exec(

    (header ?? "").trim().toLowerCase(),

  );

  if (match === null) return undefined;

  const [, version, traceId, spanId, flags] = match as unknown as [

    string,

    string,

    string,

    string,

    string,

  ];

  if (version === "ff" || traceId === ZERO_TRACE || spanId === ZERO_SPAN)

    return undefined;

  return { traceId, spanId, sampled: (Number.parseInt(flags, 16) & 1) === 1 };

}

export const formatTraceparent = (ctx: SpanContext): string =>

  `00-${ctx.traceId}-${ctx.spanId}-${ctx.sampled ? "01" : "00"}`;

// ---- spans ------------------------------------------------------------------

/** Wall-clock nanoseconds, as a decimal string (OTLP/JSON's uint64 form). */

const nowNs = (): bigint =>

  BigInt(Math.round((performance.timeOrigin + performance.now()) * 1e3)) *

  1000n;

export class Span implements SpanContext {

  readonly traceId: string;

  readonly spanId: string;

  readonly parentSpanId: string | undefined;

  readonly sampled: boolean;

  readonly startNs: bigint;

  endNs: bigint | undefined;

  readonly attributes: Attrs;

  status: { code: 0 | 1 | 2; message?: string } = { code: 0 };

  readonly events: { name: string; timeNs: bigint; attributes: Attrs }[] = [];

  constructor(

    public name: string,

    readonly kind: SpanKindName,

    parent: SpanContext | undefined,

    attributes: Attrs,

    startTime?: number,

  ) {

    // A span recorded after the fact (a run that finished) says when it began.

    this.startNs = startTime === undefined ? nowNs() : BigInt(Math.round(startTime * 1e3)) * 1000n;

    this.traceId = parent?.traceId ?? hex(16);

    this.parentSpanId = parent?.spanId;

    this.sampled = parent?.sampled ?? true;

    this.spanId = hex(8);

    this.attributes = { ...attributes };

  }

  set(key: string, value: Attr | undefined): this {

    if (value !== undefined) this.attributes[key] = value;

    return this;

  }

  event(name: string, attributes: Attrs = {}): this {

    this.events.push({ name, timeNs: nowNs(), attributes });

    return this;

  }

  /** Mark failed. An exception's message is kept; its stack is not. */

  fail(error: unknown): this {

    const message = error instanceof Error ? error.message : String(error);

    this.status = { code: 2, message };

    this.event("exception", {

      "exception.type": error instanceof Error ? error.name : typeof error,

      "exception.message": message,

    });

    return this;

  }

  end(): void {

    if (this.endNs !== undefined) return;

    this.endNs = nowNs();

    if (this.sampled && enabled()) enqueue(this);

  }

  get durationSeconds(): number {

    return Number((this.endNs ?? nowNs()) - this.startNs) / 1e9;

  }

}

const active = new AsyncLocalStorage<Span>();

/** The span the current code runs inside, if any. */

export const currentSpan = (): Span | undefined => active.getStore();

export interface SpanOptions {

  readonly kind?: SpanKindName;

  /** Parent explicitly; default is the current span. `null` = a new trace. */

  readonly parent?: SpanContext | null;

  readonly attributes?: Attrs;

  /** When it began, in epoch milliseconds, if before now. */

  readonly startTime?: number;

}

export function startSpan(name: string, options: SpanOptions = {}): Span {

  const parent =

    options.parent === null ? undefined : (options.parent ?? currentSpan());

  return new Span(name, options.kind ?? "internal", parent, options.attributes ?? {}, options.startTime);

}

/**

 * Run `fn` inside a new span: it is current for everything `fn` awaits, it

 * ends when `fn` settles, and a throw marks it failed (and still throws).

 */

export async function span<T>(

  name: string,

  fn: (span: Span) => T | Promise<T>,

  options: SpanOptions = {},

): Promise<T> {

  const s = startSpan(name, options);

  try {

    return await active.run(s, () => fn(s));

  } catch (error) {

    s.fail(error);

    throw error;

  } finally {

    s.end();

  }

}

/** Run `fn` with `s` as the current span, without ending it. */

export const withSpan = <T>(s: Span, fn: () => T): T => active.run(s, fn);

// ---- HTTP -------------------------------------------------------------------

/**

 * A server span for one request, parented on the caller's `traceparent`.

 * Records `http.server.request.duration` too. `route` names the matched

 * route template (`/notes/:id`) once it is known — set it on the span.

 */

export async function serveSpan<R extends Response | undefined>(

  request: Request,

  handle: (span: Span) => R | Promise<R>,

  attributes: Attrs = {},

): Promise<R> {

  const url = new URL(request.url);

  const s = startSpan(request.method, {

    kind: "server",

    parent: parseTraceparent(request.headers.get("traceparent")) ?? null,

    attributes: {

      "http.request.method": request.method,

      "url.path": url.pathname,

      "url.scheme":

        request.headers.get("x-forwarded-proto") ?? url.protocol.slice(0, -1),

      "server.address": (

        request.headers.get("x-forwarded-host") ??

        request.headers.get("host") ??

        url.hostname

      ).replace(/:\d+$/, ""),

      "user_agent.original": request.headers.get("user-agent") ?? undefined,

      ...attributes,

    },

  });

  let status = 500;

  try {

    const response = await active.run(s, () => handle(s));

    // undefined = the socket was upgraded (a websocket): it answered 101.

    status = response?.status ?? 101;

    return response;

  } catch (error) {

    s.fail(error);

    throw error;

  } finally {

    endServerSpan(s, status);

  }

}

/**

 * Trace a whole `Bun.serve` options object: its `fetch` runs inside

 * `serveSpan`, everything else (websocket handlers, port, error) untouched.

 * The one-line retrofit for an app that routes by hand:

 *

 *   Bun.serve(traceServe({ port, fetch(request, server) { ... } }))

 *

 * A route template isn't known here, so spans are named by method only —

 * set `http.route` on `currentSpan()` from the handler if you route by name.

 */

export function traceServe<

  T extends { fetch: (request: Request, ...rest: never[]) => unknown },

>(options: T): T {

  const inner = options.fetch;

  return {

    ...options,

    fetch(this: unknown, request: Request, ...rest: never[]) {

      return serveSpan(

        request,

        async () =>

          (await inner.call(this, request, ...rest)) as Response | undefined,

      );

    },

  };

}

/**

 * Finish a server span made by hand (a framework's request hooks): status,

 * the `METHOD /route` name, and the duration histogram — what `serveSpan`

 * does in its `finally`.

 */

export function endServerSpan(s: Span, status: number): void {

  const method = s.attributes["http.request.method"];

  s.set("http.response.status_code", status);

  if (status >= 500 && s.status.code === 0) s.status = { code: 2 };

  const route = s.attributes["http.route"];

  if (typeof route === "string" && typeof method === "string")

    s.name = `${method} ${route}`;

  s.end();

  requestDuration.record(s.durationSeconds, {

    "http.request.method": typeof method === "string" ? method : undefined,

    "http.route": typeof route === "string" ? route : undefined,

    "http.response.status_code": status,

  });

}

/**

 * `fetch` that makes a client span and hands the callee our `traceparent`,

 * so its spans join this trace. Same signature as `fetch`.

 */

export async function tracedFetch(

  input: string | URL | Request,

  init: RequestInit = {},

): Promise<Response> {

  const request =

    input instanceof Request

      ? new Request(input, init)

      : new Request(String(input), init);

  const url = new URL(request.url);

  return span(

    `${request.method}`,

    async (s) => {

      const headers = new Headers(request.headers);

      headers.set("traceparent", formatTraceparent(s));

      const response = await nativeFetch(new Request(request, { headers }));

      s.set("http.response.status_code", response.status);

      if (response.status >= 500) s.status = { code: 2 };

      return response;

    },

    {

      kind: "client",

      attributes: {

        "http.request.method": request.method,

        "server.address": url.hostname,

        "server.port": url.port === "" ? undefined : Number(url.port),

        "url.full": `${url.origin}${url.pathname}`,

      },

    },

  );

}

// ---- metrics ----------------------------------------------------------------

const DURATION_BUCKETS = [

  0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10,

];

const attrKey = (attrs: Attrs): string =>

  JSON.stringify(

    Object.entries(attrs)

      .filter(([, v]) => v !== undefined)

      .sort(([a], [b]) => (a < b ? -1 : 1)),

  );

interface Instrument {

  readonly name: string;

  readonly unit: string;

  readonly description: string;

  toOtlp(startNs: string, nowNs: string): unknown | undefined;

}

const instruments: Instrument[] = [];

const processStartNs = nowNs().toString();

export interface Histogram {

  record(value: number, attributes?: Attrs): void;

}

/** A cumulative explicit-bucket histogram. Keep attribute values few. */

export function histogram(

  name: string,

  options: { unit?: string; description?: string; buckets?: number[] } = {},

): Histogram {

  const bounds = options.buckets ?? DURATION_BUCKETS;

  const points = new Map<

    string,

    { attrs: Attrs; count: number; sum: number; buckets: number[] }

  >();

  instruments.push({

    name,

    unit: options.unit ?? "s",

    description: options.description ?? "",

    toOtlp(startNs, timeNs) {

      if (points.size === 0) return undefined;

      return {

        histogram: {

          aggregationTemporality: 2,

          dataPoints: [...points.values()].map((p) => ({

            attributes: otlpAttrs(p.attrs),

            startTimeUnixNano: startNs,

            timeUnixNano: timeNs,

            count: String(p.count),

            sum: p.sum,

            bucketCounts: p.buckets.map(String),

            explicitBounds: bounds,

          })),

        },

      };

    },

  });

  return {

    record(value, attributes = {}) {

      if (!enabled() || !Number.isFinite(value)) return;

      const key = attrKey(attributes);

      let p = points.get(key);

      if (p === undefined) {

        p = {

          attrs: attributes,

          count: 0,

          sum: 0,

          buckets: new Array(bounds.length + 1).fill(0),

        };

        points.set(key, p);

      }

      p.count += 1;

      p.sum += value;

      let i = bounds.findIndex((b) => value <= b);

      if (i === -1) i = bounds.length;

      p.buckets[i] = (p.buckets[i] ?? 0) + 1;

    },

  };

}

export interface Counter {

  add(value?: number, attributes?: Attrs): void;

}

/** A cumulative monotonic sum. */

export function counter(

  name: string,

  options: { unit?: string; description?: string } = {},

): Counter {

  const points = new Map<string, { attrs: Attrs; value: number }>();

  instruments.push({

    name,

    unit: options.unit ?? "1",

    description: options.description ?? "",

    toOtlp(startNs, timeNs) {

      if (points.size === 0) return undefined;

      return {

        sum: {

          aggregationTemporality: 2,

          isMonotonic: true,

          dataPoints: [...points.values()].map((p) => ({

            attributes: otlpAttrs(p.attrs),

            startTimeUnixNano: startNs,

            timeUnixNano: timeNs,

            asDouble: p.value,

          })),

        },

      };

    },

  });

  return {

    add(value = 1, attributes = {}) {

      if (!enabled() || !(value >= 0)) return;

      const key = attrKey(attributes);

      const p = points.get(key);

      if (p === undefined) points.set(key, { attrs: attributes, value });

      else p.value += value;

    },

  };

}

/** A value read at export time: queue depth, rows in a table, players online. */

export function gauge(

  name: string,

  read: () => number | { value: number; attributes?: Attrs }[],

  options: { unit?: string; description?: string } = {},

): void {

  instruments.push({

    name,

    unit: options.unit ?? "1",

    description: options.description ?? "",

    toOtlp(_startNs, timeNs) {

      let values: { value: number; attributes?: Attrs }[];

      try {

        const got = read();

        values = typeof got === "number" ? [{ value: got }] : got;

      } catch {

        return undefined;

      }

      if (values.length === 0) return undefined;

      return {

        gauge: {

          dataPoints: values.map((v) => ({

            attributes: otlpAttrs(v.attributes ?? {}),

            timeUnixNano: timeNs,

            asDouble: v.value,

          })),

        },

      };

    },

  });

}

/** The semantic-convention HTTP server histogram `serveSpan` records into. */

const requestDuration = histogram("http.server.request.duration", {

  unit: "s",

  description: "Duration of HTTP server requests.",

});

// ---- logs -------------------------------------------------------------------

export type Severity = "debug" | "info" | "warn" | "error";

const SEVERITY_NUMBER: Record<Severity, number> = { debug: 5, info: 9, warn: 13, error: 17 };

export interface LogOptions {

  readonly severity?: Severity;

  readonly attributes?: Attrs;

  /**

   * Who is speaking, when it is not this process: the forge files each

   * app's output under the app's own `service.name`. Added to (and wins

   * over) this process's resource.

   */

  readonly resource?: Attrs;

}

interface LogEntry {

  readonly timeNs: bigint;

  readonly body: string;

  readonly severity: Severity;

  readonly attributes: Attrs;

  readonly resource: Attrs | undefined;

  readonly traceId: string | undefined;

  readonly spanId: string | undefined;

}

const MAX_LOG_QUEUE = 4096;

let logQueue: LogEntry[] = [];

/**

 * One log record, tied to the current span when there is one. A no-op

 * without an endpoint; never throws, never blocks.

 */

export function emitLog(body: string, options: LogOptions = {}): void {

  if (!enabled()) return;

  const s = currentSpan();

  logQueue.push({

    timeNs: nowNs(),

    body,

    severity: options.severity ?? "info",

    attributes: options.attributes ?? {},

    resource: options.resource,

    traceId: s?.traceId,

    spanId: s?.spanId,

  });

  // The oldest go first: a flood keeps its newest lines.

  if (logQueue.length > MAX_LOG_QUEUE) logQueue = logQueue.slice(-MAX_LOG_QUEUE);

  startTimer();

}

/** Log records in OTLP/JSON, grouped by who spoke. */

export function logsToOtlp(entries: readonly LogEntry[]): unknown {

  const byResource = new Map<string, { resource: Attrs | undefined; records: unknown[] }>();

  for (const e of entries) {

    const key = e.resource === undefined ? "" : attrKey(e.resource);

    let group = byResource.get(key);

    if (group === undefined) byResource.set(key, (group = { resource: e.resource, records: [] }));

    group.records.push({

      timeUnixNano: e.timeNs.toString(),

      observedTimeUnixNano: e.timeNs.toString(),

      severityNumber: SEVERITY_NUMBER[e.severity],

      severityText: e.severity.toUpperCase(),

      body: { stringValue: e.body },

      attributes: otlpAttrs(e.attributes),

      ...(e.traceId === undefined ? {} : { traceId: e.traceId }),

      ...(e.spanId === undefined ? {} : { spanId: e.spanId }),

    });

  }

  return {

    resourceLogs: [...byResource.values()].map((g) => ({

      resource: resource(g.resource),

      scopeLogs: [{ scope: SCOPE, logRecords: g.records }],

    })),

  };

}

// ---- export -----------------------------------------------------------------

function otlpValue(value: Attr): Record<string, unknown> {

  if (typeof value === "boolean") return { boolValue: value };

  if (typeof value === "number")

    return Number.isInteger(value)

      ? { intValue: String(value) }

      : { doubleValue: value };

  return { stringValue: value };

}

export function otlpAttrs(attrs: Attrs): unknown[] {

  return Object.entries(attrs)

    .filter((e): e is [string, Attr] => e[1] !== undefined)

    .map(([key, value]) => ({ key, value: otlpValue(value) }));

}

const resource = (over?: Attrs) => ({

  attributes: otlpAttrs({

    "telemetry.sdk.name": "forge-otel",

    "telemetry.sdk.language": "js",

    "process.runtime.name": process.versions.bun === undefined ? "node" : "bun",

    "process.runtime.version": process.versions.bun ?? process.versions.node,

    ...config.resource,

    "service.name": config.service,

    ...over,

  }),

});

const SCOPE = { name: "forge-otel", version: "1" };

/** One span in OTLP/JSON (ids are lowercase hex, times decimal ns). */

export function spanToOtlp(s: Span): unknown {

  return {

    traceId: s.traceId,

    spanId: s.spanId,

    ...(s.parentSpanId === undefined ? {} : { parentSpanId: s.parentSpanId }),

    name: s.name,

    kind: SpanKind[s.kind],

    startTimeUnixNano: s.startNs.toString(),

    endTimeUnixNano: (s.endNs ?? s.startNs).toString(),

    attributes: otlpAttrs(s.attributes),

    events: s.events.map((e) => ({

      name: e.name,

      timeUnixNano: e.timeNs.toString(),

      attributes: otlpAttrs(e.attributes),

    })),

    status: s.status,

  };

}

const MAX_QUEUE = 2048;

const BATCH = 512;

let queue: Span[] = [];

let timer: ReturnType<typeof setInterval> | undefined;

let lastMetrics = 0;

function enqueue(s: Span): void {

  queue.push(s);

  if (queue.length > MAX_QUEUE) queue = queue.slice(-MAX_QUEUE);

  startTimer();

}

function startTimer(): void {

  if (timer === undefined) {

    timer = setInterval(() => void flush(), 5_000);

    // Telemetry must never be the reason a process stays alive.

    (timer as { unref?: () => void }).unref?.();

  }

}

async function post(path: string, body: unknown): Promise<boolean> {

  if (config.endpoint === undefined) return false;

  try {

    const response = await nativeFetch(`${config.endpoint}${path}`, {

      method: "POST",

      headers: { "content-type": "application/json" },

      body: JSON.stringify(body),

      signal: AbortSignal.timeout(5_000),

    });

    return response.ok;

  } catch {

    return false;

  }

}

/** Send what is queued now. Resolves when sent (or dropped); never throws. */

export async function flush(): Promise<void> {

  while (queue.length > 0) {

    const batch = queue.splice(0, BATCH);

    await post("/v1/traces", {

      resourceSpans: [

        { resource: resource(), scopeSpans: [{ scope: SCOPE, spans: batch.map(spanToOtlp) }] },

      ],

    });

  }

  while (logQueue.length > 0) {

    const batch = logQueue.splice(0, BATCH);

    await post("/v1/logs", logsToOtlp(batch));

  }

  if (Date.now() - lastMetrics >= 30_000) {

    lastMetrics = Date.now();

    await flushMetrics();

  }

}

export function metricsToOtlp(): unknown | undefined {

  const timeNs = nowNs().toString();

  const metrics = instruments.flatMap((i) => {

    const data = i.toOtlp(processStartNs, timeNs);

    return data === undefined

      ? []

      : [{ name: i.name, unit: i.unit, description: i.description, ...(data as object) }];

  });

  if (metrics.length === 0) return undefined;

  return {

    resourceMetrics: [

      { resource: resource(), scopeMetrics: [{ scope: SCOPE, metrics }] },

    ],

  };

}

export async function flushMetrics(): Promise<void> {

  if (!enabled()) return;

  const body = metricsToOtlp();

  if (body !== undefined) await post("/v1/metrics", body);

}

/** Start the periodic exporter even before the first span (metrics-only apps). */

export function start(): void {

  if (!enabled() || timer !== undefined) return;

  timer = setInterval(() => void flush(), 5_000);

  (timer as { unref?: () => void }).unref?.();

}