WRNexusJS/packages/tracking/src/pipeline.ts

export type TelemetryKind = "error" | "event" | "metric" | "span";

export interface TelemetryEnvelope {
  id: string;
  kind: TelemetryKind;
  name: string;
  timestamp: number;
  traceId?: string;
  userId?: string;
  tenantId?: string;
  attributes: Record<string, unknown>;
  payload?: unknown;
}

export interface TelemetrySink {
  name?: string;
  send(events: TelemetryEnvelope[]): void | Promise<void>;
}

export interface TelemetryPipeline {
  emit(
    input: Omit<TelemetryEnvelope, "id" | "timestamp"> &
      Partial<Pick<TelemetryEnvelope, "id" | "timestamp">>,
  ): Promise<void>;
  flush(): Promise<void>;
  close(): Promise<void>;
  size(): number;
}

export interface TelemetryPipelineOptions {
  sinks: TelemetrySink[];
  batchSize?: number;
  flushIntervalMs?: number;
  maxQueueSize?: number;
  sampleRate?: number;
  beforeSend?: (event: TelemetryEnvelope) => TelemetryEnvelope | null;
  onSinkError?: (
    error: unknown,
    sink: TelemetrySink,
    events: readonly TelemetryEnvelope[],
  ) => void | Promise<void>;
  now?: () => number;
  random?: () => number;
}

function positiveInteger(value: number, label: string): number {
  if (!Number.isInteger(value) || value < 1) {
    throw new RangeError(`${label} must be a positive integer`);
  }
  return value;
}

export function createTelemetryPipeline(options: TelemetryPipelineOptions): TelemetryPipeline {
  if (!options.sinks.length) {
    throw new Error("WRN-TRACKING-NO-SINKS: at least one telemetry sink is required");
  }

  const batchSize = positiveInteger(options.batchSize ?? 20, "batchSize");
  const maxQueueSize = positiveInteger(options.maxQueueSize ?? 1_000, "maxQueueSize");
  const flushIntervalMs = options.flushIntervalMs ?? 5_000;
  if (!Number.isFinite(flushIntervalMs) || flushIntervalMs < 0) {
    throw new RangeError("flushIntervalMs must be a non-negative number");
  }

  const sampleRate = Math.min(1, Math.max(0, options.sampleRate ?? 1));
  const now = options.now ?? Date.now;
  const random = options.random ?? Math.random;
  const queue: TelemetryEnvelope[] = [];
  let timer: ReturnType<typeof setInterval> | null = null;
  let flushing: Promise<void> | null = null;

  function prependWithinLimit(events: readonly TelemetryEnvelope[]): void {
    queue.unshift(...events);
    if (queue.length > maxQueueSize) {
      queue.splice(maxQueueSize);
    }
  }

  const flush = async (): Promise<void> => {
    if (flushing) return flushing;

    const batch = queue.splice(0, batchSize);
    if (!batch.length) return;

    flushing = (async () => {
      let failed = false;
      await Promise.all(
        options.sinks.map(async (sink) => {
          try {
            await sink.send(batch);
          } catch (error) {
            failed = true;
            await options.onSinkError?.(error, sink, batch);
          }
        }),
      );

      // At-least-once delivery: retry the batch once regardless of how many
      // sinks failed. Requeueing per sink would duplicate the same batch.
      if (failed) prependWithinLimit(batch);
    })().finally(() => {
      flushing = null;
    });

    await flushing;
  };

  if (flushIntervalMs > 0) {
    timer = setInterval(() => void flush(), flushIntervalMs);
  }

  return {
    async emit(input) {
      if (sampleRate < 1 && random() > sampleRate) return;

      let event: TelemetryEnvelope | null = {
        id: input.id ?? crypto.randomUUID(),
        timestamp: input.timestamp ?? now(),
        kind: input.kind,
        name: input.name,
        traceId: input.traceId,
        userId: input.userId,
        tenantId: input.tenantId,
        attributes: input.attributes ?? {},
        payload: input.payload,
      };
      event = options.beforeSend?.(event) ?? event;
      if (!event) return;

      if (queue.length >= maxQueueSize) queue.shift();
      queue.push(event);
      if (queue.length >= batchSize) await flush();
    },

    flush,

    async close() {
      if (timer) clearInterval(timer);
      timer = null;

      while (queue.length) {
        const before = queue.length;
        await flush();
        if (queue.length >= before) break;
      }
    },

    size: () => queue.length,
  };
}

export const telemetryConsoleSink: TelemetrySink = {
  name: "console",
  send(events) {
    for (const event of events) {
      console.log(`[telemetry:${event.kind}] ${event.name}`, event.attributes);
    }
  },
};