import type {
  PacingPermit,
  PacingRule,
  RateStateBackend,
} from './rate-state-backend';

/**
 * In-memory Rate State Backend for the single-process `cjs_node20` runner.
 *
 * In one Node process the local buckets ARE the global state, so this is the
 * correct, zero-round-trip backend. The window/concurrency math is lifted from
 * the previous `PlayRateLimitScheduler` (sliding window + per-rule concurrency +
 * serialized acquire), plus a `penalize` cooldown the scheduler lacked so the
 * in-process runner also honors a server-observed Retry-After.
 */

const MIN_CONCURRENCY_WAIT_MS = 10;

interface RuleState {
  windowStartedAt: number;
  startedInWindow: number;
  activeCount: number;
}

interface Options {
  now?: () => number;
  sleep?: (ms: number) => Promise<void>;
}

export class InMemoryRateStateBackend implements RateStateBackend {
  private readonly ruleStates = new Map<string, RuleState>();
  private readonly cooldownUntilByBucket = new Map<string, number>();
  private lock: Promise<void> = Promise.resolve();
  private readonly now: () => number;
  private readonly sleep: (ms: number) => Promise<void>;

  constructor(options: Options = {}) {
    this.now = options.now ?? (() => Date.now());
    this.sleep =
      options.sleep ??
      ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms)));
  }

  async acquire(input: {
    bucketId: string;
    rules: readonly PacingRule[];
    signal?: AbortSignal;
  }): Promise<PacingPermit> {
    const { bucketId, rules, signal } = input;
    if (rules.length === 0) {
      return { release() {} };
    }
    // Stable order to avoid deadlock between concurrent acquirers on shared rules.
    const ordered = [...rules].sort((a, b) => a.ruleId.localeCompare(b.ruleId));

    while (true) {
      if (signal?.aborted) {
        throw signal.reason instanceof Error
          ? signal.reason
          : new Error('Rate-state acquire aborted.');
      }
      const decision = await this.withLock(() => {
        const now = this.now();
        let waitMs = 0;

        const cooldownUntil = this.cooldownUntilByBucket.get(bucketId) ?? 0;
        if (cooldownUntil > now) waitMs = Math.max(waitMs, cooldownUntil - now);

        for (const rule of ordered) {
          const state = this.getRuleState(bucketId, rule, now);
          this.resetExpiredWindow(state, rule.windowMs, now);
          if (state.startedInWindow >= rule.requestsPerWindow) {
            waitMs = Math.max(waitMs, state.windowStartedAt + rule.windowMs - now);
          }
          if (rule.maxConcurrency != null && state.activeCount >= rule.maxConcurrency) {
            waitMs = Math.max(waitMs, MIN_CONCURRENCY_WAIT_MS);
          }
        }

        if (waitMs > 0) return { acquired: false, waitMs: Math.max(1, waitMs) } as const;

        for (const rule of ordered) {
          const state = this.getRuleState(bucketId, rule, now);
          state.startedInWindow += 1;
          if (rule.maxConcurrency != null) state.activeCount += 1;
        }
        return { acquired: true, waitMs: 0 } as const;
      });

      if (decision.acquired) {
        let released = false;
        return {
          // The concurrency decrement is intentionally not awaitable (the permit
          // signature is sync `release(): void`). It still runs serialized under
          // the lock; do not "fix" this by trying to await a void.
          release: () => {
            if (released) return;
            released = true;
            void this.withLock(() => {
              for (const rule of ordered) {
                if (rule.maxConcurrency == null) continue;
                const state = this.ruleStates.get(this.key(bucketId, rule.ruleId));
                if (state && state.activeCount > 0) state.activeCount -= 1;
              }
            });
          },
        };
      }
      await this.sleep(decision.waitMs);
    }
  }

  penalize(input: { bucketId: string; cooldownMs: number }): void {
    if (input.cooldownMs <= 0) return;
    const until = this.now() + input.cooldownMs;
    const existing = this.cooldownUntilByBucket.get(input.bucketId) ?? 0;
    this.cooldownUntilByBucket.set(input.bucketId, Math.max(existing, until));
  }

  private key(bucketId: string, ruleId: string): string {
    return `${bucketId}::${ruleId}`;
  }

  private getRuleState(
    bucketId: string,
    rule: PacingRule,
    now: number,
  ): RuleState {
    const key = this.key(bucketId, rule.ruleId);
    const existing = this.ruleStates.get(key);
    if (existing) return existing;
    const created: RuleState = {
      windowStartedAt: now,
      startedInWindow: 0,
      activeCount: 0,
    };
    this.ruleStates.set(key, created);
    return created;
  }

  private resetExpiredWindow(
    state: RuleState,
    windowMs: number,
    now: number,
  ): void {
    if (windowMs <= 0) {
      state.windowStartedAt = now;
      state.startedInWindow = 0;
      return;
    }
    if (now - state.windowStartedAt < windowMs) return;
    const elapsed = Math.floor((now - state.windowStartedAt) / windowMs);
    state.windowStartedAt += elapsed * windowMs;
    state.startedInWindow = 0;
    if (now - state.windowStartedAt >= windowMs) state.windowStartedAt = now;
  }

  private async withLock<T>(fn: () => T | Promise<T>): Promise<T> {
    const previous = this.lock;
    let releaseLock!: () => void;
    this.lock = new Promise<void>((resolve) => {
      releaseLock = resolve;
    });
    await previous;
    try {
      return await fn();
    } finally {
      releaseLock();
    }
  }
}