tech-log-frontend/src/adapters/telemetry/best-effort-telemetry.ts

import { projectTelemetryEvent } from "../../contracts/telemetry.ts";
import { queueSizeBucket } from "../../contracts/diagnostic-buckets.ts";
import type {
  TelemetryEvent,
  TelemetryEventName,
} from "../../contracts/telemetry.ts";
import type { TelemetryPort } from "../../application/ports/telemetry-port.ts";
import { assertBoundedCapacity } from "../platform/bounded-capacity.ts";

export type TelemetryAdapter = TelemetryPort &
  Readonly<{
    flush(): Promise<void>;
    pendingCount(): number;
    droppedCount(): number;
    dropReasons(): Readonly<Record<string, number>>;
    deliveryEvidence(): TelemetryEvent | null;
    dispose(): void;
  }>;

export type TelemetryAdapterOptions = Readonly<{
  enabled: boolean;
  endpoint?: string;
  fetcher?: typeof fetch;
  maxQueue?: number;
  schedule?: (callback: () => void) => void;
  now?: () => number;
  onDrop?: (event: TelemetryEvent) => void;
  lifecycle?: Pick<EventTarget, "addEventListener" | "removeEventListener">;
}>;

export const noOpTelemetry: TelemetryAdapter = Object.freeze({
  emit: () => {},
  flush: async () => {},
  pendingCount: () => 0,
  droppedCount: () => 0,
  dropReasons: () => Object.freeze({}),
  deliveryEvidence: () => null,
  dispose: () => {},
});

/**
 * N-04. Disposal is terminal: there is no durable queue and no resurrection.
 */
type TelemetryLifecycle = "ACTIVE" | "DISPOSED";

/** N-11. Documented absolute ceiling for the in-memory best-effort queue. */
export const MAX_TELEMETRY_QUEUE = 10_000;

export { assertBoundedCapacity } from "../platform/bounded-capacity.ts";


export function createTelemetryAdapter(
  options: TelemetryAdapterOptions,
): TelemetryAdapter {
  if (!options.enabled || !options.endpoint) {
    return noOpTelemetry;
  }

  const endpoint = options.endpoint;
  const fetcher = options.fetcher ?? fetch;
  const maxQueue = assertBoundedCapacity(
    options.maxQueue ?? 100,
    MAX_TELEMETRY_QUEUE,
    "telemetry maxQueue",
  );
  const schedule = options.schedule ?? queueMicrotask;
  const queue: TelemetryEvent[] = [];
  let lifecycleState: TelemetryLifecycle = "ACTIVE";
  /** Scheduled callbacks captured before disposal must not run afterwards. */
  let scheduleGeneration = 0;
  let scheduled = false;
  let activeFlush: Promise<void> | null = null;
  let activeSink: AbortController | null = null;
  let dropped = 0;
  const dropReasons = new Map<string, number>();
  let lastDeliveryEvidence: TelemetryEvent | null = null;
  const lifecycle =
    options.lifecycle ??
    (typeof globalThis.addEventListener === "function" &&
    typeof globalThis.removeEventListener === "function"
      ? globalThis
      : undefined);

  function recordDrop(reason: string, count = 1): void {
    const safeReason =
      {
        "queue-full": "queue-full",
        "sink-failure": "sink-failure",
        "serialization-failure": "serialization-failure",
        "unknown-attributes": "invalid-context",
        "invalid-attribute-value": "invalid-context",
        "missing-required-attributes": "invalid-context",
        "unregistered-event": "invalid-event",
      }[reason] ?? "invalid-event";
    dropped += count;
    dropReasons.set(safeReason, (dropReasons.get(safeReason) ?? 0) + count);
    const internal = projectTelemetryEvent(
      "telemetry.delivery.dropped",
      {
        reason: safeReason,
        queue_size_bucket: queueSizeBucket(queue.length),
      },
      options.now,
    );
    if (internal.success) {
      lastDeliveryEvidence = internal.event;
      try {
        options.onDrop?.(internal.event);
      } catch {
        // Drop observers are deliberately nonrecursive.
      }
    }
  }

  function scheduleFlush(): void {
    if (scheduled || lifecycleState === "DISPOSED") return;
    scheduled = true;
    const generation = scheduleGeneration;
    schedule(() => {
      // A callback captured before disposal belongs to a dead generation.
      if (generation !== scheduleGeneration) return;
      scheduled = false;
      void flush();
    });
  }

  function emit(
    eventName: TelemetryEventName,
    attributes: Record<string, unknown>,
  ): void {
    if (lifecycleState === "DISPOSED") return;
    const projected = projectTelemetryEvent(
      eventName,
      attributes,
      options.now,
    );
    if (!projected.success) {
      recordDrop(projected.reason);
      return;
    }

    if (queue.length >= maxQueue) {
      queue.shift();
      recordDrop("queue-full");
    }
    queue.push(projected.event);

    scheduleFlush();
  }

  /**
   * `flush()` joins the active delivery instead of resolving immediately, so an
   * awaited flush really means "the in-flight batch has settled".
   */
  function flush(): Promise<void> {
    if (activeFlush) return activeFlush;
    if (lifecycleState === "DISPOSED" || queue.length === 0) {
      return Promise.resolve();
    }
    const generation = scheduleGeneration;
    const run = async () => {
      const batch = queue.splice(0, queue.length);
      const controller = new AbortController();
      activeSink = controller;
      try {
        const response = await fetcher(endpoint, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ events: batch }),
          keepalive: true,
          signal: controller.signal,
        });
        // A sink that ignored the abort must not update post-dispose state.
        if (generation !== scheduleGeneration) return;
        if (!response.ok) recordDrop("sink-failure", batch.length);
      } catch {
        if (generation !== scheduleGeneration) return;
        recordDrop("sink-failure", batch.length);
      } finally {
        if (activeSink === controller) activeSink = null;
        activeFlush = null;
        if (generation === scheduleGeneration && queue.length > 0) {
          scheduleFlush();
        }
      }
    };
    activeFlush = run();
    return activeFlush;
  }

  const flushBeforePageExit = () => {
    void flush();
  };
  lifecycle?.addEventListener("pagehide", flushBeforePageExit);

  /**
   * N-04. Terminal disposal: one state transition, no further admission, no
   * further scheduling, and no recursive drop telemetry while shutting down.
   */
  function dispose(): void {
    if (lifecycleState === "DISPOSED") return;
    lifecycleState = "DISPOSED";
    scheduleGeneration += 1;
    scheduled = false;
    lifecycle?.removeEventListener("pagehide", flushBeforePageExit);
    queue.length = 0;
    activeSink?.abort();
    activeSink = null;
    activeFlush = null;
  }

  return Object.freeze({
    emit,
    flush,
    pendingCount: () => queue.length,
    droppedCount: () => dropped,
    dropReasons: () => Object.freeze(Object.fromEntries(dropReasons)),
    deliveryEvidence: () =>
      lastDeliveryEvidence ? structuredClone(lastDeliveryEvidence) : null,
    dispose,
  });
}

/**
 * Propagates only a structurally valid W3C traceparent. Invalid/raw headers are
 * discarded rather than logged or surfaced.
 *
 */
export function safeTraceparent(
  traceparent: string | null | undefined,
): string | null {
  return typeof traceparent === "string" &&
    /^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/i.test(traceparent)
    ? traceparent.toLowerCase()
    : null;
}