export type GatewayPostgresLane = 'ordinary' | 'commit';

export const GATEWAY_POSTGRES_MAX_ORDINARY_ACTIVE = 8;
export const GATEWAY_POSTGRES_MAX_TOTAL_ACTIVE = 10;
export const GATEWAY_POSTGRES_MAX_CONSECUTIVE_COMMIT_PRIORITY_GRANTS = 8;

export type GatewayPostgresAdmissionSnapshot = {
  active: Record<GatewayPostgresLane, number> & { total: number };
  waiting: Record<GatewayPostgresLane, number>;
  cancelled: number;
  commitPriorityGrants: number;
};

export class GatewayPostgresAdmissionCancelledError extends Error {
  constructor() {
    super('Gateway Postgres admission was cancelled.');
    this.name = 'GatewayPostgresAdmissionCancelledError';
  }
}

type Waiter = {
  lane: GatewayPostgresLane;
  signal: AbortSignal | null;
  onAbort: (() => void) | null;
  resolve: (release: () => void) => void;
  reject: (error: Error) => void;
  settled: boolean;
};

export class GatewayPostgresAdmission {
  readonly #waiting: Record<GatewayPostgresLane, Waiter[]> = {
    ordinary: [],
    commit: [],
  };
  #ordinaryActive = 0;
  #commitActive = 0;
  #cancelled = 0;
  #commitPriorityGrants = 0;
  #consecutiveCommitPriorityGrants = 0;

  constructor(
    private readonly limits: {
      maxOrdinaryActive?: number;
      maxTotalActive?: number;
    } = {},
  ) {}

  async acquire(
    lane: GatewayPostgresLane,
    options: { signal?: AbortSignal } = {},
  ): Promise<() => void> {
    if (options.signal?.aborted) {
      this.#cancelled += 1;
      throw new GatewayPostgresAdmissionCancelledError();
    }
    if (this.#canGrant(lane) && this.#waiting.commit.length === 0) {
      return this.#grant(lane);
    }

    return await new Promise<() => void>((resolve, reject) => {
      const waiter: Waiter = {
        lane,
        signal: options.signal ?? null,
        onAbort: null,
        resolve,
        reject,
        settled: false,
      };
      waiter.onAbort = () => {
        if (!this.#remove(waiter)) return;
        this.#cancelled += 1;
        reject(new GatewayPostgresAdmissionCancelledError());
      };
      waiter.signal?.addEventListener('abort', waiter.onAbort, { once: true });
      this.#waiting[lane].push(waiter);
      if (waiter.signal?.aborted) {
        waiter.onAbort();
        return;
      }
      this.#drain();
    });
  }

  snapshot(): GatewayPostgresAdmissionSnapshot {
    return {
      active: {
        ordinary: this.#ordinaryActive,
        commit: this.#commitActive,
        total: this.#activeTotal(),
      },
      waiting: {
        ordinary: this.#waiting.ordinary.length,
        commit: this.#waiting.commit.length,
      },
      cancelled: this.#cancelled,
      commitPriorityGrants: this.#commitPriorityGrants,
    };
  }

  #activeTotal(): number {
    return this.#ordinaryActive + this.#commitActive;
  }

  #canGrant(lane: GatewayPostgresLane): boolean {
    if (this.#activeTotal() >= this.#maxTotalActive()) return false;
    return (
      lane === 'commit' ||
      this.#ordinaryActive < this.#maxOrdinaryActive()
    );
  }

  #maxTotalActive(): number {
    return this.limits.maxTotalActive ?? GATEWAY_POSTGRES_MAX_TOTAL_ACTIVE;
  }

  #maxOrdinaryActive(): number {
    return (
      this.limits.maxOrdinaryActive ?? GATEWAY_POSTGRES_MAX_ORDINARY_ACTIVE
    );
  }

  #grant(lane: GatewayPostgresLane): () => void {
    if (lane === 'ordinary') this.#ordinaryActive += 1;
    else this.#commitActive += 1;
    let released = false;
    return () => {
      if (released) return;
      released = true;
      if (lane === 'ordinary') this.#ordinaryActive -= 1;
      else this.#commitActive -= 1;
      this.#drain();
    };
  }

  #drain(): void {
    while (this.#activeTotal() < this.#maxTotalActive()) {
      const ordinaryCanRun =
        this.#waiting.ordinary.length > 0 &&
        this.#ordinaryActive < this.#maxOrdinaryActive();
      if (
        ordinaryCanRun &&
        this.#waiting.commit.length > 0 &&
        this.#consecutiveCommitPriorityGrants >=
          GATEWAY_POSTGRES_MAX_CONSECUTIVE_COMMIT_PRIORITY_GRANTS
      ) {
        this.#consecutiveCommitPriorityGrants = 0;
        this.#resolve(this.#waiting.ordinary.shift()!);
        continue;
      }
      const commit = this.#waiting.commit.shift();
      if (commit) {
        if (ordinaryCanRun) {
          this.#commitPriorityGrants += 1;
          this.#consecutiveCommitPriorityGrants += 1;
        } else {
          this.#consecutiveCommitPriorityGrants = 0;
        }
        this.#resolve(commit);
        continue;
      }
      if (ordinaryCanRun) {
        this.#consecutiveCommitPriorityGrants = 0;
        this.#resolve(this.#waiting.ordinary.shift()!);
        continue;
      }
      if (this.#waiting.ordinary.length === 0) {
        this.#consecutiveCommitPriorityGrants = 0;
      }
      return;
    }
  }

  #resolve(waiter: Waiter): void {
    waiter.settled = true;
    this.#cleanup(waiter);
    waiter.resolve(this.#grant(waiter.lane));
  }

  #remove(waiter: Waiter): boolean {
    if (waiter.settled) return false;
    const queue = this.#waiting[waiter.lane];
    const index = queue.indexOf(waiter);
    if (index < 0) return false;
    waiter.settled = true;
    queue.splice(index, 1);
    this.#cleanup(waiter);
    return true;
  }

  #cleanup(waiter: Waiter): void {
    if (waiter.signal && waiter.onAbort) {
      waiter.signal.removeEventListener('abort', waiter.onAbort);
    }
  }
}