/**
 * @description 从数组创建一个立即推送全部值的 `ReadableStream`。
 *
 * @example
 *   ```
 *   // Expect: [1, 2, 3]
 *   const example1 = await Array.fromAsync(streamFromArrayEager([1, 2, 3]))
 *   // Expect: []
 *   const example2 = await Array.fromAsync(streamFromArrayEager([]))
 *   ```
 */
export const streamFromArrayEager = <T>(values: T[]): ReadableStream<T> => {
  const stream = new ReadableStream<T>({
    start(controller): void {
      for (const value of values) {
        controller.enqueue(value)
      }
      controller.close()
    },
  })
  return stream
}

/**
 * @description 从数组创建一个按需逐项提供值的 `ReadableStream`。
 *
 * @example
 *   ```
 *   // Expect: [1, 2, 3]
 *   const example1 = await Array.fromAsync(streamFromArrayLazy([1, 2, 3]))
 *   // Expect: []
 *   const example2 = await Array.fromAsync(streamFromArrayLazy([]))
 *   ```
 */
export const streamFromArrayLazy = <T>(values: T[]): ReadableStream<T> => {
  let i = 0
  const stream = new ReadableStream<T>({
    pull(controller): void {
      if (i < values.length) {
        controller.enqueue(values[i] as T)
        i = i + 1
      } else {
        controller.close()
      }
    },
  })
  return stream
}

/**
 * @description `streamConsumeInMicroTask` 的配置项。
 */
export interface StreamConsumeInMicroTaskOptions<T> {
  readableStream: ReadableStream<T>
  onValue?: ((chunk: T) => void | Promise<void>) | undefined
  onDone?: (() => void | Promise<void>) | undefined
  onError?: ((error: Error) => void | Promise<void>) | undefined
}

/**
 * @description 在微任务队列中消费 `ReadableStream`。
 *
 * @example
 *   ```
 *   const values: number[] = []
 *   streamConsumeInMicroTask({
 *     readableStream: streamFromArrayEager([1, 2]),
 *     onValue: (value) => {
 *       values.push(value)
 *     },
 *   })
 *   // Expect: [1, 2]
 *   const example1 = await Promise.resolve().then(() => Promise.resolve(values))
 *   ```
 */
export const streamConsumeInMicroTask = <T>(options: StreamConsumeInMicroTaskOptions<T>): void => {
  const { readableStream, onValue, onDone, onError } = options

  const reader = readableStream.getReader()

  const read = (): void => {
    queueMicrotask(() => {
      void reader
        .read()
        .then(async ({ done, value }) => {
          if (done) {
            await onDone?.()
            return
          }
          await onValue?.(value)
          read()
        })
        .catch(async (reason: unknown) => {
          console.error(`Error reading stream: ${String(reason)}`)
          await onError?.(new Error(String(reason)))
        })
    })
  }

  read()
}

/**
 * @description `streamConsumeInMacroTask` 的配置项。
 */
export interface StreamConsumeInMacroTaskOptions<T> {
  readableStream: ReadableStream<T>
  onValue?: ((chunk: T) => void | Promise<void>) | undefined
  onDone?: (() => void | Promise<void>) | undefined
  onError?: ((error: Error) => void | Promise<void>) | undefined
}

/**
 * @description 在宏任务队列中消费 `ReadableStream`。
 *
 * @example
 *   ```
 *   const values: number[] = []
 *   await new Promise<void>((resolve) => {
 *     streamConsumeInMacroTask({
 *       readableStream: streamFromArrayEager([1, 2]),
 *       onValue: (value) => {
 *         values.push(value)
 *       },
 *       onDone: resolve,
 *     })
 *   })
 *   // Expect: [1, 2]
 *   const example1 = values
 *   ```
 */
export const streamConsumeInMacroTask = <T>(options: StreamConsumeInMacroTaskOptions<T>): void => {
  const { readableStream, onValue, onDone, onError } = options

  const reader = readableStream.getReader()
  const read = (): void => {
    setTimeout(() => {
      void reader
        .read()
        .then(async (chunk) => {
          const { done, value } = chunk
          if (done === true) {
            await onDone?.()
            return
          }
          await onValue?.(value)
          read()
        })
        .catch(async (reason: unknown) => {
          console.error(`Error reading stream: ${String(reason)}`)
          await onError?.(new Error(String(reason)))
        })
    }, 0)
  }
  read()
}

/**
 * @description `streamTransformInMacroTask` 的配置项。
 */
export interface StreamTransformInMacroTaskOptions<T, U> {
  readableStream?: ReadableStream<T> | undefined
  reader?: ReadableStreamDefaultReader<T> | undefined
  onChunk?:
    | ((
        chunk: ReadableStreamReadResult<T>,
        controller: ReadableStreamDefaultController<U>,
      ) => void | Promise<void>)
    | undefined
  onError?: ((error: Error) => void | Error | Promise<void | Error>) | undefined
}

/**
 * @description 在宏任务队列中把一个 `ReadableStream` 转换为另一个 `ReadableStream`。
 *
 * @example
 *   ```
 *   const source = streamFromArrayEager([1, 2, 3])
 *   const transformed = streamTransformInMacroTask({
 *     readableStream: source,
 *     onChunk: (chunk, controller) => {
 *       if (chunk.done === true) {
 *         controller.close()
 *         return
 *       }
 *       controller.enqueue(chunk.value * 10)
 *     },
 *   })
 *   // Expect: [10, 20, 30]
 *   const example1 = await Array.fromAsync(transformed)
 *   ```
 */
export const streamTransformInMacroTask = <T, U>(
  options: StreamTransformInMacroTaskOptions<T, U>,
): ReadableStream<U> => {
  const { readableStream, reader, onChunk, onError } = options

  if (reader === undefined && readableStream === undefined) {
    throw new Error("Either readableStream or reader must be provided")
  }

  const preparedReader = reader ?? readableStream!.getReader()
  const stream = new ReadableStream<U>({
    start(controller): void {
      const read = (): void => {
        setTimeout(() => {
          void preparedReader
            .read()
            .then(async (chunk) => {
              let shouldContinue = true
              const proxyController: ReadableStreamController<U> = {
                // oxlint-disable-next-line no-misused-spread
                ...controller,
                enqueue: (chunk: U): void => {
                  controller.enqueue(chunk)
                  shouldContinue = true
                },
                close: (): void => {
                  controller.close()
                  shouldContinue = false
                },
                error: (error: Error): void => {
                  controller.error(error)
                  shouldContinue = false
                },
              }
              // @ts-expect-error Bun 的 ReadableStreamDefaultReader.read() 返回 ReadableStreamDefaultReadResult，
              // done 分支的 value 被标成可选；这里对外使用的是 W3C/lib.dom 的 ReadableStreamReadResult，
              // 运行时结构兼容，但两套声明在 TypeScript 中不兼容。
              await onChunk?.(chunk, proxyController)
              if (shouldContinue === true) {
                read()
              }
            })
            .catch(async (reason: unknown) => {
              const error = new Error(String(reason))
              const refinedError = await onError?.(error)
              controller.error(refinedError ?? error)
            })
        }, 0)
      }
      read()
    },
  })

  return stream
}