import type { ClientOptions } from "openai"
import { OpenAI } from "openai"
import type { Stream } from "openai/streaming"
import { Agent, setGlobalDispatcher } from "undici"

import type { WithAbortSignal } from "#Source/abort/index.ts"
import * as Aio from "#Source/aio/index.ts"
import {
  scheduleMacroTask,
  streamConsumeInMacroTask,
  streamTransformInMacroTask,
} from "#Source/basic/index.ts"
import type { LoggerFriendly, LoggerFriendlyOptions } from "#Source/log/index.ts"
import { Logger } from "#Source/log/index.ts"
import type { Either } from "#Source/result/index.ts"
import { controllerFromEitherType } from "#Source/result/index.ts"
import { Tube } from "#Source/tube/index.ts"

export { OpenAI } from "openai"
export { Stream } from "openai/streaming"

export interface WithOpenai {
  openai: Openai
}

export interface OriginalChatCompletionOptions
  extends Omit<OpenAI.Chat.ChatCompletionCreateParams, never>, WithAbortSignal {}
export type OriginalChatCompletionResult = OpenAI.Chat.ChatCompletion
export type OriginalChatCompletionChunkResult = OpenAI.Chat.Completions.ChatCompletionChunk

export interface CustomChatCompletionOptions extends OriginalChatCompletionOptions {
  /**
   * @description In milliseconds.
   */
  timeout?: number | undefined
}
export interface CustomChatCompletionNonStreamingLeft {
  code: "NO_CHOICE" | "TOO_LONG" | "FILTERED" | "REFUSED" | "EMPTY" | "UNKNOWN"
}
export interface CustomChatCompletionNonStreamingRight {
  content: string
  token: number
}
export type CustomChatCompletionNonStreamingResult = Either<
  CustomChatCompletionNonStreamingLeft,
  CustomChatCompletionNonStreamingRight
>

export interface CustomChatCompletionStreamingLeft {
  code: never
}
export interface Completion {
  content: Aio.TextContent
  token: Aio.NumberContent
}
export interface CustomChatCompletionStreamingRight {
  completionTube: Tube<Completion>
}
export type CustomChatCompletionStreamingResult = Either<
  CustomChatCompletionStreamingLeft,
  CustomChatCompletionStreamingRight
>

export interface OriginalEmbeddingOptions
  extends OpenAI.Embeddings.EmbeddingCreateParams, WithAbortSignal {}
export type OriginalEmbeddingResult = OpenAI.Embeddings.CreateEmbeddingResponse

export interface CustomEmbeddingOptions extends WithAbortSignal {
  input: string
  model?: OpenAI.Embeddings.EmbeddingCreateParams["model"] | undefined
  dimensions?: OpenAI.Embeddings.EmbeddingCreateParams["dimensions"] | undefined
}
export interface CustomEmbeddingResultLeft {
  code: "NO_EMBEDDING" | "MODEL_MISMATCH" | "UNKNOWN"
}
export interface CustomEmbeddingResultRight {
  embedding: number[]
  usage: {
    promptTokens: number
    totalTokens: number
  }
}
export type CustomEmbeddingResult = Either<CustomEmbeddingResultLeft, CustomEmbeddingResultRight>

export type CustomSpeechOptions = Omit<
  OpenAI.Audio.SpeechCreateParams,
  "input" | "model" | "voice" | "response_format"
> & {
  input: string
}

export interface OpenaiOptions extends LoggerFriendlyOptions {
  apiKey: Exclude<ClientOptions["apiKey"], undefined | null>
  baseUrl: Exclude<ClientOptions["baseURL"], undefined | null>
  /**
   * @description In milliseconds.
   *
   * @default 5 * 60 * 1000
   */
  timeout?: ClientOptions["timeout"] | undefined
}
interface ResolvedOpenaiOptions {
  apiKey: Exclude<ClientOptions["apiKey"], undefined | null>
  baseUrl: Exclude<ClientOptions["baseURL"], undefined | null>
  timeout: Exclude<ClientOptions["timeout"], undefined | null>
}
export class Openai implements LoggerFriendly {
  protected options: ResolvedOpenaiOptions

  readonly logger: Logger
  protected openai: OpenAI

  constructor(options: OpenaiOptions) {
    this.options = {
      apiKey: options.apiKey,
      baseUrl: options.baseUrl,
      timeout: options.timeout ?? 5 * 60 * 1_000,
    }

    this.logger = Logger.fromOptions(options).setDefaultName("Openai")
    this.openai = new OpenAI({
      baseURL: this.options.baseUrl,
      // NOTE: 若服务方证书出现问题可暂时开启这个选项，问题解决后应该第一时间关闭。
      // fetch: this.customFetchThatIgnoreTheCAs,
      timeout: this.options.timeout,
      apiKey: this.options.apiKey,
    })
  }

  protected async customFetchThatIgnoreTheCAs(url: string, init?: RequestInit): Promise<Response> {
    setGlobalDispatcher(new Agent({ connect: { rejectUnauthorized: false } }))

    const response = await fetch(url, init)
    if (response.ok === false) {
      throw new Error(`Request failed with status ${response.status}`)
    }

    return response
  }

  getBaseUrl(): string {
    return this.openai.baseURL
  }

  /**
   * @description 原始 chatCompletion 方法的包装，但将其固定为非流式且不自动重试。
   */
  async originalChatCompletionNonStreaming(
    options: OriginalChatCompletionOptions,
  ): Promise<OriginalChatCompletionResult> {
    try {
      const { abortSignal, ...restOptions } = options
      const response = await this.openai.chat.completions.create(
        {
          ...restOptions,
          stream: false,
        },
        {
          maxRetries: 0,
          signal: abortSignal,
        },
      )
      return response
    } catch (exception) {
      this.logger.error("Error creating chat completion: ", exception)
      throw exception
    }
  }

  /**
   * @description 原始 chatCompletion 方法的包装，但将其固定为流式且不自动重试。
   */
  async originalChatCompletionStreaming(
    options: OriginalChatCompletionOptions,
  ): Promise<Stream<OriginalChatCompletionChunkResult>> {
    try {
      const { abortSignal, ...restOptions } = options
      const response = await this.openai.chat.completions.create(
        {
          ...restOptions,
          stream: true,
        },
        {
          maxRetries: 0,
          signal: abortSignal,
        },
      )
      return response
    } catch (exception) {
      this.logger.error("Error creating chat completion: ", exception)
      throw exception
    }
  }

  /**
   * @description 此方法不会自动重试，调用方需要通过错误信息自行处理。
   *
   * {@link https://platform.openai.com/docs/api-reference/chat | Chat - OpenAI API}
   */
  async customChatCompletionNonStreaming(
    options: CustomChatCompletionOptions,
  ): Promise<CustomChatCompletionNonStreamingResult> {
    const controller = controllerFromEitherType<CustomChatCompletionNonStreamingResult>()
    try {
      const { abortSignal, ...restOptions } = options
      const response = await this.openai.chat.completions.create(
        {
          ...restOptions,
          stream: false,
        },
        {
          maxRetries: 0,
          signal: abortSignal,
        },
      )
      const choice = response.choices[0]
      if (choice === undefined) {
        this.logger.error("Chat completion is empty.")
        return await controller.returnLeft({ code: "NO_CHOICE" })
      }

      const finishReason = choice.finish_reason
      if (finishReason === "length") {
        this.logger.error("Chat completion is too long.")
        return await controller.returnLeft({ code: "TOO_LONG" })
      }
      if (finishReason === "content_filter") {
        this.logger.error("Chat completion is filtered.")
        return await controller.returnLeft({ code: "FILTERED" })
      }

      const refusal = choice.message.refusal
      if (refusal !== null && refusal !== undefined) {
        this.logger.error(`Chat completion is refused: ${refusal}`)
        return await controller.returnLeft({ code: "REFUSED" })
      }
      const content = choice.message.content
      if (content === null) {
        this.logger.error("Chat completion is empty.")
        return await controller.returnLeft({ code: "EMPTY" })
      }

      return await controller.returnRight({
        content,
        // TODO: count the tokens
        token: 0,
      })
    } catch (exception) {
      this.logger.error("Error creating chat completion: ", exception)
      return await controller.returnLeft({ code: "UNKNOWN" })
    }
  }

  async customChatCompletionStreaming(
    options: CustomChatCompletionOptions,
  ): Promise<CustomChatCompletionStreamingResult> {
    const controller = controllerFromEitherType<CustomChatCompletionStreamingResult>()

    const { timeout = this.options.timeout, abortSignal } = options
    const openaiClient = this.openai
    const logger = this.logger

    const completionTube = new Tube<Completion>({
      historyCount: Infinity,
      replayHistory: true,
    })

    const start = async (): Promise<void> => {
      const { abortSignal: _, ...restOptions } = options
      try {
        // Pass the AbortSignal to the OpenAI request to support aborting
        logger.log("Starting chat completion streaming.")
        const response = await openaiClient.chat.completions.create(
          {
            ...restOptions,
            stream: true,
          },
          {
            timeout,
            maxRetries: 0,
            signal: abortSignal, // Include the abort signal in the request options
          },
        )
        logger.log("Chat completion streaming started.")

        // NOTE: 以下将 response 转换成 ReadableStream
        const readableStream = streamTransformInMacroTask<
          Uint8Array,
          OriginalChatCompletionChunkResult
        >({
          reader: response.toReadableStream().getReader(),
          onChunk: (chunk, controller) => {
            if (abortSignal !== undefined && abortSignal.aborted) {
              logger.log("Original chat completion api's stream aborted by invoker.")
              const error = new Error("Original chat completion api's stream aborted by invoker.")
              controller.error(error)
              return
            }
            if (chunk.done) {
              logger.log("Original chat completion api's stream ended.")
              controller.close()
              return
            }
            const textDecoder = new TextDecoder()
            const decoded = textDecoder.decode(chunk.value)
            const parsed = JSON.parse(decoded) as OriginalChatCompletionChunkResult
            // logger.log("Original chat completion api's stream chunk:", JSON.stringify(parsed))
            controller.enqueue(parsed)
            return
          },
          onError: (error) => {
            logger.error("Error reading original chat completion api's stream:", error)
          },
        })

        const latestCompletion: Completion = {
          content: {
            deltaList: [],
            total: "",
          },
          token: {
            deltaList: [],
            total: 0,
          },
        }

        streamConsumeInMacroTask({
          readableStream,
          onValue: async (chunk) => {
            // If the signal is aborted, break and stop the stream
            if (abortSignal !== undefined && abortSignal.aborted) {
              throw new Error("Stream aborted by abort signal")
            }
            const choice = chunk.choices[0]
            if (choice === undefined) {
              // NOTE: sometimes there will be no choice in the chunk, just skip the chunk
              // Example chunk:
              // {
              //   "id": "chatcmpl-hTMAcJUVRpO5Etj64tdowhgpQalcP",
              //   "object": "chat.completion.chunk",
              //   "created": 1734481734,
              //   "model": "gpt-4o-mini",
              //   "choices": [],
              //   "usage": {
              //     "prompt_tokens": 2345,
              //     "completion_tokens": 90,
              //     "total_tokens": 2435,
              //     "prompt_tokens_details": {
              //       "cached_tokens": 0,
              //       "audio_tokens": 0
              //     },
              //     "completion_tokens_details": {
              //       "audio_tokens": 0
              //     }
              //   }
              // }
              return
            }

            const finishReason = choice.finish_reason
            if (finishReason === "stop") {
              // the model hit a natural stop point or a provided stop sequence, do nothing
            }
            if (finishReason === "length") {
              // the maximum number of tokens specified in the request was reached
              logger.error("Chat completion is too long.")
            }
            if (finishReason === "content_filter") {
              logger.error("Chat completion is filtered.")
              throw new Error("Chat completion is filtered.")
            }

            const refusal = choice.delta.refusal
            if (refusal !== null && refusal !== undefined) {
              logger.error(`Chat completion is refused: ${refusal}`)
              throw new Error(`Chat completion is refused: ${refusal}`)
            }

            const deltaContent = choice.delta.content
            if (deltaContent !== null && deltaContent !== undefined) {
              latestCompletion.content = Aio.applyDeltaToTextContent(latestCompletion.content, {
                type: "append",
                text: deltaContent,
              })
              latestCompletion.token = Aio.applyDeltaToNumberContent(latestCompletion.token, {
                type: "append",
                value: 0,
              })
              await completionTube.pushData(structuredClone(latestCompletion))
            }
          },
          onDone: async () => {
            await completionTube.end()
          },
          onError: async (error) => {
            await completionTube.pushError(error)
          },
        })
      } catch (exception) {
        if (abortSignal !== undefined && abortSignal.aborted) {
          logger.log("Original chat completion api's stream aborted by invoker.")
          const error = new Error("Original chat completion api's stream aborted by invoker.")
          await completionTube.pushError(error)
        } else {
          logger.error("Error starting chat completion streaming: ", exception)
          const error = new Error(`Error starting chat completion streaming, ${String(exception)}`)
          await completionTube.pushError(error)
        }
      }
    }
    scheduleMacroTask({
      task: async () => {
        await start()
      },
    })

    return await controller.returnRight({
      completionTube,
    })
  }

  /**
   * @description 原始 embedding 方法的包装，将其最大重试次数固定为 0。
   */
  async originalEmbedding(options: OriginalEmbeddingOptions): Promise<OriginalEmbeddingResult> {
    const { abortSignal, ...restOptions } = options
    try {
      const response = await this.openai.embeddings.create(
        {
          ...restOptions,
        },
        {
          maxRetries: 0,
          signal: abortSignal,
        },
      )
      return response
    } catch (exception) {
      this.logger.error("Error creating embedding: ", exception)
      throw exception
    }
  }

  /**
   * @description {@link https://platform.openai.com/docs/api-reference/embeddings | Embeddings - OpenAI API}
   */
  async customEmbedding(options: CustomEmbeddingOptions): Promise<CustomEmbeddingResult> {
    const controller = controllerFromEitherType<CustomEmbeddingResult>()
    const logger = this.logger
    try {
      const {
        abortSignal,
        input,
        model = "text-embedding-3-large",
        // IMPORTANT: the number of dimensions should be 1536 currently, ask @kongxiangyan for more information.
        dimensions = 1_536,
      } = options
      logger.log(
        `Creating embedding with model: ${model}, dimensions: ${dimensions}, input: ${input}`,
      )
      const response = await this.openai.embeddings.create(
        {
          input,
          model,
          dimensions,
        },
        {
          maxRetries: 0,
          signal: abortSignal,
        },
      )
      logger.log("Get embedding response.")
      // logger.log("Embedding response: ", JSON.stringify(response, null, 2))
      if (response.data.length === 0) {
        logger.error("Embedding is empty.")
        return await controller.returnLeft({ code: "NO_EMBEDDING" })
      }
      if (response.model !== model) {
        logger.error(`Model mismatch: ${response.model} !== ${model}`)
        return await controller.returnLeft({ code: "MODEL_MISMATCH" })
      }

      const embedding = response.data[0]!.embedding
      const promptTokens = response.usage.prompt_tokens
      const totalTokens = response.usage.total_tokens
      return await controller.returnRight({
        embedding,
        usage: {
          promptTokens,
          totalTokens,
        },
      })
    } catch (exception) {
      logger.error("Error creating embedding: ", exception)
      return await controller.returnLeft({ code: "UNKNOWN" })
    }
  }

  /**
   * @description {@link https://platform.openai.com/docs/api-reference/audio/createSpeech | Audio - OpenAI API}
   */
  async customSpeech(options: CustomSpeechOptions, stream?: boolean): Promise<Buffer> {
    try {
      const response = await this.openai.audio.speech.create(
        {
          ...options,
          model: "tts-1",
          voice: "alloy",
          response_format: "opus",
        },
        {
          stream: stream ?? false,
        },
      )
      const buffer = Buffer.from(await response.arrayBuffer())
      return buffer
    } catch (exception) {
      this.logger.error("Error creating speech: ", exception)
      throw exception
    }
  }
}