import type {
  InternalWaitConstraintsEntry,
  InternalWaitConstraintsOptions,
} from "./internal/wait-constraints.ts"
import {
  internalAssertTimeoutOption,
  internalBindWaitConstraints,
  internalCleanupWaitConstraints,
  internalThrowIfAborted,
} from "./internal/wait-constraints.ts"
import type { InternalWaitQueueEntry } from "./internal/wait-queue.ts"
import { InternalWaitQueue } from "./internal/wait-queue.ts"
import type { CoordinationPermitDetails } from "./permit.ts"
import { Permit } from "./permit.ts"

const internalAssertMaxConcurrency = (maxConcurrency: number): void => {
  if (
    Number.isFinite(maxConcurrency) === false ||
    Number.isInteger(maxConcurrency) === false ||
    maxConcurrency <= 0
  ) {
    throw new RangeError("Semaphore maxConcurrency must be a finite integer greater than 0.")
  }
}

/**
 * @description 表示 `Semaphore` 成功获取后的占用句柄。
 */
export type SemaphorePermit = Permit<CoordinationPermitDetails<"semaphore">>

/**
 * @description 表示信号量等待队列中的单个节点。
 */
interface InternalSemaphoreQueueEntry
  extends InternalWaitQueueEntry<InternalSemaphoreQueueEntry>, InternalWaitConstraintsEntry {
  resolve: (permit: SemaphorePermit) => void
  reject: (reason: unknown) => void
  isSettled: boolean
}

/**
 * @description 表示调用 `Semaphore.acquire()` 时可选的等待约束。
 */
export interface SemaphoreAcquireOptions extends InternalWaitConstraintsOptions {
  /**
   * @description 单次等待允许持续的最长时间，单位为毫秒。
   */
  timeout?: number | undefined

  /**
   * @description 用于主动中止本次等待的信号。
   */
  abortSignal?: AbortSignal | undefined
}

const INTERNAL_SEMAPHORE_DUPLICATE_RELEASE_MESSAGE =
  "Semaphore permit release was called more than once."

/**
 * @description 表示信号量在检测到重复释放时使用的处理函数。
 */
export type SemaphoreDuplicateReleaseHandler = (message: string) => void

/**
 * @description 表示构造 `Semaphore` 时支持的配置。
 */
export interface SemaphoreOptions {
  /**
   * @description 重复调用同一个 permit 的 `release()` 时触发的可选处理器。
   */
  onDuplicateRelease?: SemaphoreDuplicateReleaseHandler | undefined
}

/**
 * @description 表示用于限制并发进入数量的信号量。
 *
 * @example
 *   ```
 *   const semaphore = new Semaphore(2)
 *   let activeCount = 0
 *   let peakCount = 0
 *
 *   await Promise.all([
 *     semaphore.runExclusive(async () => {
 *       activeCount = activeCount + 1
 *       peakCount = Math.max(peakCount, activeCount)
 *       await Promise.resolve()
 *       activeCount = activeCount - 1
 *     }),
 *     semaphore.runExclusive(async () => {
 *       activeCount = activeCount + 1
 *       peakCount = Math.max(peakCount, activeCount)
 *       await Promise.resolve()
 *       activeCount = activeCount - 1
 *     }),
 *     semaphore.runExclusive(async () => {
 *       activeCount = activeCount + 1
 *       peakCount = Math.max(peakCount, activeCount)
 *       await Promise.resolve()
 *       activeCount = activeCount - 1
 *     }),
 *   ])
 *
 *   // Expect: 2
 *   const example1 = peakCount
 *   ```
 */
export class Semaphore {
  private readonly queue: InternalWaitQueue<InternalSemaphoreQueueEntry>

  private internalActiveCount: number

  private readonly maxConcurrency: number

  private readonly onDuplicateRelease: SemaphoreDuplicateReleaseHandler | undefined

  /**
   * @description 使用给定最大并发数创建信号量。
   */
  constructor(maxConcurrency: number, options: SemaphoreOptions = {}) {
    internalAssertMaxConcurrency(maxConcurrency)

    this.queue = new InternalWaitQueue<InternalSemaphoreQueueEntry>()
    this.internalActiveCount = 0
    this.maxConcurrency = maxConcurrency
    this.onDuplicateRelease = options.onDuplicateRelease
  }

  /**
   * @description 判断当前是否已经达到并发上限。
   */
  isSaturated(): boolean {
    return this.internalActiveCount >= this.maxConcurrency
  }

  /**
   * @description 读取当前正在占用中的数量。
   */
  getActiveCount(): number {
    return this.internalActiveCount
  }

  /**
   * @description 读取当前还剩多少可立即分配的许可。
   */
  getAvailableCount(): number {
    return this.maxConcurrency - this.internalActiveCount
  }

  /**
   * @description 读取当前等待中的任务数量。
   */
  getPendingCount(): number {
    return this.queue.getCount()
  }

  /**
   * @description 读取信号量的最大并发容量。
   */
  getMaxConcurrency(): number {
    return this.maxConcurrency
  }

  /**
   * @description 在仍有可用容量时，按 FIFO 顺序持续放行等待者。
   */
  private dispatch(): void {
    while (this.internalActiveCount < this.maxConcurrency && this.queue.isEmpty() === false) {
      const nextEntry = this.queue.shift()
      if (nextEntry === undefined) {
        return
      }

      if (nextEntry.isSettled === true) {
        continue
      }

      this.internalActiveCount = this.internalActiveCount + 1
      nextEntry.isSettled = true
      internalCleanupWaitConstraints(nextEntry)
      nextEntry.resolve(this.buildPermit())
    }
  }

  /**
   * @description 构造一个信号量 permit，并在释放时归还一个并发槽位。
   */
  private buildPermit(): SemaphorePermit {
    return new Permit<CoordinationPermitDetails<"semaphore">>({
      details: { coordination: "semaphore" },
      duplicateReleaseMessage: INTERNAL_SEMAPHORE_DUPLICATE_RELEASE_MESSAGE,
      onDuplicateRelease: this.onDuplicateRelease,
      onRelease: (): void => {
        this.internalActiveCount = this.internalActiveCount - 1
        this.dispatch()
      },
    })
  }

  /**
   * @description 以失败状态结束一个等待节点，并尝试继续调度后继等待者。
   */
  private rejectEntry(entry: InternalSemaphoreQueueEntry, reason: unknown): void {
    if (entry.isSettled === true) {
      return
    }

    entry.isSettled = true
    internalCleanupWaitConstraints(entry)
    this.queue.remove(entry)
    entry.reject(reason)
    this.dispatch()
  }

  /**
   * @description 尝试立即获取一个许可；若没有剩余容量或前面已有排队任务，则返回 `undefined`。
   */
  tryAcquire(): SemaphorePermit | undefined {
    if (this.internalActiveCount >= this.maxConcurrency || this.queue.isEmpty() === false) {
      return undefined
    }

    this.internalActiveCount = this.internalActiveCount + 1
    return this.buildPermit()
  }

  /**
   * @description 等待直到成功获取一个许可。
   */
  async acquire(options: SemaphoreAcquireOptions = {}): Promise<SemaphorePermit> {
    internalAssertTimeoutOption("Semaphore acquire", options.timeout)

    internalThrowIfAborted("Semaphore acquire", options.abortSignal)

    const immediatePermit = this.tryAcquire()
    if (immediatePermit !== undefined) {
      return immediatePermit
    }

    const permit = await new Promise<SemaphorePermit>((resolve, reject) => {
      const entry: InternalSemaphoreQueueEntry = {
        resolve,
        reject,
        isSettled: false,
        isQueued: false,
        previous: undefined,
        next: undefined,
        timeoutId: undefined,
        abortSignal: undefined,
        abortListener: undefined,
      }

      internalBindWaitConstraints(entry, options, {
        abortOperation: "Semaphore acquire",
        timeoutOperation: "Semaphore acquire",
        onAbort: (error) => {
          this.rejectEntry(entry, error)
        },
        onTimeout: (error) => {
          this.rejectEntry(entry, error)
        },
      })
      this.queue.enqueue(entry)
      this.dispatch()
    })

    return permit
  }

  /**
   * @description 在信号量许可保护下执行回调。
   */
  async runExclusive<T>(callback: () => Promise<T>, options?: SemaphoreAcquireOptions): Promise<T>
  async runExclusive<T>(callback: () => T, options?: SemaphoreAcquireOptions): Promise<T>
  async runExclusive<T>(
    callback: () => Promise<T> | T,
    options: SemaphoreAcquireOptions = {},
  ): Promise<T> {
    const permit = await this.acquire(options)
    try {
      return await callback()
    } finally {
      permit.release()
    }
  }
}