import { CancelledError, OperationTimeoutError, fileErrorFromPayload } from '../schemas/errors.js';
import type { RpcError, RpcResponse } from '../schemas/envelope.js';

interface PendingEntry {
  resolve: (result: unknown) => void;
  reject: (err: unknown) => void;
  timer: ReturnType<typeof setTimeout> | null;
  abortListener: (() => void) | null;
  signal?: AbortSignal | undefined;
}

export interface RegisterOptions {
  timeoutMs?: number;
  signal?: AbortSignal;
  /** Called when client cancels (signal aborted or timeout); transport can ship an `RpcCancel`. */
  onCancel?: (reason: 'timeout' | 'signal') => void;
}

/**
 * Client-side tracker for in-flight RPC calls. Routes incoming
 * `RpcResponse`/`RpcError` to the right pending promise via correlation id.
 */
export class PendingRpcRegistry {
  private readonly entries = new Map<string, PendingEntry>();

  /** Register a pending request. Returns a Promise resolved/rejected by routing. */
  register<T = unknown>(requestId: string, opts: RegisterOptions = {}): Promise<T> {
    const p = new Promise<T>((resolve, reject) => {
      const cleanup = (): void => {
        const e = this.entries.get(requestId);
        if (e === undefined) return;
        if (e.timer !== null) clearTimeout(e.timer);
        if (e.abortListener !== null && e.signal !== undefined) {
          e.signal.removeEventListener('abort', e.abortListener);
        }
        this.entries.delete(requestId);
      };

      const entry: PendingEntry = {
        resolve: (v) => {
          cleanup();
          resolve(v as T);
        },
        reject: (err) => {
          cleanup();
          reject(err);
        },
        timer: null,
        abortListener: null,
        ...(opts.signal !== undefined ? { signal: opts.signal } : {}),
      };

      if (opts.timeoutMs !== undefined && opts.timeoutMs > 0) {
        entry.timer = setTimeout(() => {
          opts.onCancel?.('timeout');
          entry.reject(new OperationTimeoutError());
        }, opts.timeoutMs);
      }

      if (opts.signal !== undefined) {
        if (opts.signal.aborted) {
          // Defer microtask so caller's await sees the rejection.
          queueMicrotask(() => {
            opts.onCancel?.('signal');
            entry.reject(new CancelledError());
          });
        } else {
          const listener = (): void => {
            opts.onCancel?.('signal');
            entry.reject(new CancelledError());
          };
          entry.abortListener = listener;
          opts.signal.addEventListener('abort', listener, { once: true });
        }
      }

      this.entries.set(requestId, entry);
    });
    // Attach a no-op rejection handler so Bun's strict unhandled-rejection
    // detection doesn't flag the registered promise BEFORE the caller's
    // `await` attaches its handler. The caller's await still observes the
    // rejection through its own handler chain.
    p.catch(() => {
      /* suppress unhandled-rejection warning */
    });
    return p;
  }

  /** Route an incoming response. */
  resolveResponse(env: RpcResponse): void {
    const entry = this.entries.get(env.id);
    if (entry === undefined) return; // unknown id — possibly a stale duplicate
    entry.resolve(env.result);
  }

  /** Route an incoming error envelope. */
  resolveError(env: RpcError): void {
    const entry = this.entries.get(env.id);
    if (entry === undefined) return;
    entry.reject(fileErrorFromPayload(env.error));
  }

  /** Reject all pending entries — for shutdown. */
  rejectAll(reason: unknown): void {
    for (const entry of this.entries.values()) entry.reject(reason);
    this.entries.clear();
  }

  /** For tests / metrics. */
  size(): number {
    return this.entries.size;
  }
}
