WRNexusJS/packages/core/src/resilience.ts

export type Duration = number | `${number}${"ms" | "s" | "m" | "h"}`;

export type BackoffStrategy = "fixed" | "exponential" | ((attempt: number) => Duration);

export interface CircuitBreakerOptions {
  failures: number;
  resetAfter: Duration;
  successesToClose?: number;
}

export interface CircuitBreakerSnapshot {
  state: "closed" | "open" | "half-open";
  failures: number;
  successes: number;
  retryAfterMs: number;
}

export class ResilienceError extends Error {
  constructor(
    public readonly code:
      | "WRN-RESILIENCE-TIMEOUT"
      | "WRN-RESILIENCE-ABORTED"
      | "WRN-RESILIENCE-CIRCUIT-OPEN"
      | "WRN-RESILIENCE-BULKHEAD-FULL",
    message: string,
    options?: ErrorOptions,
  ) {
    super(message, options);
    this.name = "ResilienceError";
  }
}

export function durationMs(value: Duration): number {
  if (typeof value === "number") {
    if (!Number.isFinite(value) || value < 0)
      throw new TypeError("Duration must be finite and non-negative.");
    return value;
  }
  const match = /^(\d+(?:\.\d+)?)(ms|s|m|h)$/.exec(value);
  if (!match) throw new TypeError(`Invalid duration: ${value}`);
  const scale = { ms: 1, s: 1_000, m: 60_000, h: 3_600_000 }[match[2]!]!;
  return Number(match[1]) * scale;
}

export class CircuitBreaker {
  private failures = 0;
  private successes = 0;
  private openedAt = 0;
  private probing = false;

  constructor(private readonly options: CircuitBreakerOptions) {
    if (!Number.isInteger(options.failures) || options.failures < 1) {
      throw new TypeError("Circuit breaker failures must be a positive integer.");
    }
    durationMs(options.resetAfter);
  }

  snapshot(now = Date.now()): CircuitBreakerSnapshot {
    const resetAfter = durationMs(this.options.resetAfter);
    const elapsed = now - this.openedAt;
    const open = this.openedAt > 0 && elapsed < resetAfter;
    return {
      state: open ? "open" : this.openedAt > 0 ? "half-open" : "closed",
      failures: this.failures,
      successes: this.successes,
      retryAfterMs: open ? Math.max(0, resetAfter - elapsed) : 0,
    };
  }

  async execute<T>(operation: () => Promise<T>): Promise<T> {
    const health = this.snapshot();
    if (health.state === "open" || (health.state === "half-open" && this.probing)) {
      throw new ResilienceError(
        "WRN-RESILIENCE-CIRCUIT-OPEN",
        `Circuit is open; retry after ${health.retryAfterMs}ms.`,
      );
    }
    if (health.state === "half-open") this.probing = true;
    try {
      const value = await operation();
      this.failures = 0;
      this.successes += 1;
      if (this.successes >= (this.options.successesToClose ?? 1)) this.openedAt = 0;
      return value;
    } catch (error) {
      this.successes = 0;
      this.failures += 1;
      if (this.failures >= this.options.failures) this.openedAt = Date.now();
      throw error;
    } finally {
      this.probing = false;
    }
  }
}

export interface BulkheadOptions {
  concurrency: number;
  queue?: number;
}

export class Bulkhead {
  private active = 0;
  private readonly waiting: Array<() => void> = [];

  constructor(private readonly options: BulkheadOptions) {
    if (!Number.isInteger(options.concurrency) || options.concurrency < 1) {
      throw new TypeError("Bulkhead concurrency must be a positive integer.");
    }
    if (options.queue !== undefined && (!Number.isInteger(options.queue) || options.queue < 0)) {
      throw new TypeError("Bulkhead queue must be a non-negative integer.");
    }
  }

  get snapshot(): Readonly<{ active: number; queued: number; capacity: number }> {
    return { active: this.active, queued: this.waiting.length, capacity: this.options.concurrency };
  }

  async execute<T>(operation: () => Promise<T>): Promise<T> {
    if (this.active >= this.options.concurrency) {
      if (this.waiting.length >= (this.options.queue ?? 0)) {
        throw new ResilienceError(
          "WRN-RESILIENCE-BULKHEAD-FULL",
          "Bulkhead capacity is exhausted.",
        );
      }
      await new Promise<void>((resolve) => this.waiting.push(resolve));
    }
    this.active += 1;
    try {
      return await operation();
    } finally {
      this.active -= 1;
      this.waiting.shift()?.();
    }
  }
}

export interface ResilientCallOptions<T> {
  run: (signal: AbortSignal, attempt: number) => Promise<T>;
  timeout?: Duration;
  retries?: number;
  retryDelay?: Duration;
  backoff?: BackoffStrategy;
  circuitBreaker?: CircuitBreaker | CircuitBreakerOptions;
  bulkhead?: Bulkhead | BulkheadOptions;
  signal?: AbortSignal;
  retryWhen?: (error: unknown, attempt: number) => boolean | Promise<boolean>;
  fallback?: (error: unknown, signal: AbortSignal) => T | Promise<T>;
  onRetry?: (error: unknown, attempt: number, delayMs: number) => void;
}

const breakerInstances = new WeakMap<CircuitBreakerOptions, CircuitBreaker>();
const bulkheadInstances = new WeakMap<BulkheadOptions, Bulkhead>();

function breakerFor(value: CircuitBreaker | CircuitBreakerOptions): CircuitBreaker {
  if (value instanceof CircuitBreaker) return value;
  const existing = breakerInstances.get(value);
  if (existing) return existing;
  const created = new CircuitBreaker(value);
  breakerInstances.set(value, created);
  return created;
}

function bulkheadFor(value: Bulkhead | BulkheadOptions): Bulkhead {
  if (value instanceof Bulkhead) return value;
  const existing = bulkheadInstances.get(value);
  if (existing) return existing;
  const created = new Bulkhead(value);
  bulkheadInstances.set(value, created);
  return created;
}

async function abortable<T>(operation: Promise<T>, signal: AbortSignal): Promise<T> {
  if (signal.aborted) throw signal.reason;
  let cleanup = () => {};
  const aborted = new Promise<never>((_resolve, reject) => {
    const listener = () => reject(signal.reason);
    signal.addEventListener("abort", listener, { once: true });
    cleanup = () => signal.removeEventListener("abort", listener);
  });
  try {
    return await Promise.race([operation, aborted]);
  } finally {
    cleanup();
  }
}

function abortError(signal: AbortSignal): ResilienceError {
  return new ResilienceError("WRN-RESILIENCE-ABORTED", "Resilient call was aborted.", {
    cause: signal.reason,
  });
}

async function delay(ms: number, signal?: AbortSignal): Promise<void> {
  if (signal?.aborted) throw abortError(signal);
  await new Promise<void>((resolve, reject) => {
    const timer = setTimeout(resolve, ms);
    signal?.addEventListener(
      "abort",
      () => {
        clearTimeout(timer);
        reject(abortError(signal));
      },
      { once: true },
    );
  });
}

export async function resilientCall<T>(options: ResilientCallOptions<T>): Promise<T> {
  const retries = options.retries ?? 0;
  if (!Number.isInteger(retries) || retries < 0)
    throw new TypeError("Retries must be a non-negative integer.");
  const breaker = options.circuitBreaker ? breakerFor(options.circuitBreaker) : undefined;
  const bulkhead = options.bulkhead ? bulkheadFor(options.bulkhead) : undefined;
  const invoke = async (): Promise<T> => {
    let lastError: unknown;
    for (let attempt = 1; attempt <= retries + 1; attempt += 1) {
      if (options.signal?.aborted) throw abortError(options.signal);
      const controller = new AbortController();
      const forwardAbort = () => controller.abort(options.signal?.reason);
      options.signal?.addEventListener("abort", forwardAbort, { once: true });
      const timeout = options.timeout === undefined ? undefined : durationMs(options.timeout);
      const timer =
        timeout === undefined ? undefined : setTimeout(() => controller.abort("timeout"), timeout);
      try {
        const run = () => abortable(options.run(controller.signal, attempt), controller.signal);
        return await (breaker ? breaker.execute(run) : run());
      } catch (caught) {
        lastError =
          controller.signal.aborted && !options.signal?.aborted
            ? new ResilienceError(
                "WRN-RESILIENCE-TIMEOUT",
                `Operation timed out after ${timeout}ms.`,
                { cause: caught },
              )
            : caught;
        if (attempt > retries || !(await (options.retryWhen?.(lastError, attempt) ?? true))) break;
        const base = durationMs(options.retryDelay ?? 100);
        const wait =
          typeof options.backoff === "function"
            ? durationMs(options.backoff(attempt))
            : options.backoff === "exponential"
              ? base * 2 ** (attempt - 1)
              : base;
        options.onRetry?.(lastError, attempt, wait);
        await delay(wait, options.signal);
      } finally {
        if (timer !== undefined) clearTimeout(timer);
        options.signal?.removeEventListener("abort", forwardAbort);
      }
    }
    if (options.fallback)
      return options.fallback(lastError, options.signal ?? new AbortController().signal);
    throw lastError;
  };
  return bulkhead ? bulkhead.execute(invoke) : invoke();
}