import { RPC_ERROR_CODES, failure } from "./errors.ts";
import type { ServiceResult } from "./types.ts";

export interface RpcTarget {
  app: string;
  service: string;
  procedure: string;
}

export interface CallOptions {
  signal?: AbortSignal;
  identity?: string;
  /** Supplied from the declared procedure; only these calls may be retried. */
  idempotent?: boolean;
}

export interface Transport {
  call(target: RpcTarget, payload: unknown, options: CallOptions): Promise<ServiceResult>;
}

export interface RetryTransportOptions {
  /** Retries after the initial attempt. Default: 2. */
  retries?: number;
  /** Initial exponential-backoff delay in milliseconds. Default: 50. */
  backoffMs?: number;
  /** Consecutive retryable failures before the target circuit opens. Default: 3. */
  circuitFailureThreshold?: number;
  /** How long an open circuit rejects calls before one probe is allowed. Default: 5s. */
  circuitCooldownMs?: number;
  now?: () => number;
  sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;
}

function targetKey(target: RpcTarget): string {
  return `${target.app}/${target.service}/${target.procedure}`;
}

function defaultSleep(ms: number, signal?: AbortSignal): Promise<void> {
  return new Promise((resolve) => {
    const timer = setTimeout(resolve, ms);
    signal?.addEventListener(
      "abort",
      () => {
        clearTimeout(timer);
        resolve();
      },
      { once: true },
    );
  });
}

/**
 * Add bounded retry and a per-procedure circuit breaker to any transport.
 * The idempotency bit comes from the contract and is not caller-controlled.
 */
export function retryingTransport(base: Transport, options: RetryTransportOptions = {}): Transport {
  const retries = options.retries ?? 2;
  const backoffMs = options.backoffMs ?? 50;
  const threshold = options.circuitFailureThreshold ?? 3;
  const cooldownMs = options.circuitCooldownMs ?? 5_000;
  const now = options.now ?? Date.now;
  const sleep = options.sleep ?? defaultSleep;
  if (!Number.isInteger(retries) || retries < 0)
    throw new RangeError("rpc retries must be a non-negative integer");
  if (!Number.isFinite(backoffMs) || backoffMs < 0)
    throw new RangeError("rpc backoffMs must be non-negative");
  if (!Number.isInteger(threshold) || threshold < 1)
    throw new RangeError("rpc circuitFailureThreshold must be positive");
  if (!Number.isFinite(cooldownMs) || cooldownMs < 1)
    throw new RangeError("rpc circuitCooldownMs must be positive");

  const circuits = new Map<string, { failures: number; openUntil: number }>();
  return {
    async call(target, payload, callOptions) {
      const key = targetKey(target);
      const circuit = circuits.get(key);
      if (circuit && circuit.openUntil > now()) {
        return failure(RPC_ERROR_CODES.transport, "Service temporarily unavailable");
      }
      if (circuit?.openUntil) circuits.delete(key); // cooldown: allow one probe

      const attempts = callOptions.idempotent ? retries + 1 : 1;
      let result: ServiceResult = failure(RPC_ERROR_CODES.transport, "Service unreachable");
      for (let attempt = 0; attempt < attempts; attempt++) {
        if (callOptions.signal?.aborted) return failure(RPC_ERROR_CODES.transport, "Call aborted");
        result = await base.call(target, payload, callOptions);
        if (result.ok || !result.retryable) {
          if (result.ok) circuits.delete(key);
          return result;
        }
        if (attempt + 1 < attempts) await sleep(backoffMs * 2 ** attempt, callOptions.signal);
      }
      const failures = (circuits.get(key)?.failures ?? 0) + 1;
      circuits.set(key, {
        failures,
        openUntil: failures >= threshold ? now() + cooldownMs : 0,
      });
      return result;
    },
  };
}

export type InProcessHandler = (
  payload: unknown,
  identity?: string,
) => Promise<ServiceResult> | ServiceResult;

/** Direct transport for tests and local integration harnesses. */
export function inProcessTransport(handlers: Record<string, InProcessHandler>): Transport {
  return {
    async call(target, payload, options) {
      // Checked only at entry: this in-process transport does no I/O, so
      // nothing yields between here and the handler call below, and a signal
      // that aborts mid-flight is never observed. In-process tests therefore
      // cannot exercise a mid-call timeout path — that needs a real transport.
      if (options.signal?.aborted) return failure(RPC_ERROR_CODES.transport, "Call aborted");
      const key = `${target.service}/${target.procedure}`;
      // Object.hasOwn, not plain indexing: a prototype-inherited key (e.g.
      // "constructor/toString") would otherwise resolve to a function that
      // is not one of our handlers.
      const handler = Object.hasOwn(handlers, key) ? handlers[key] : undefined;
      if (!handler) return failure(RPC_ERROR_CODES.unknown, "Unknown procedure");
      try {
        return await handler(payload, options.identity);
      } catch {
        return failure(RPC_ERROR_CODES.handler, "Internal error");
      }
    },
  };
}