import { describe, expect } from "bun:test"
import { Effect, Layer, Stream } from "effect"
import { Base64 } from "effect/encoding"
import { Speech, SpeechClient, SpeechEvent } from "../src/experimental.js"
import { Cartesia, Deepgram, ElevenLabs, Google, OpenAI } from "../src/providers.js"
import { it } from "./lib/effect.js"
import { dynamicResponse } from "./lib/http.js"
import { sseEvents } from "./lib/sse.js"

// Recorded tests in test/provider cover lowering and decoding; these cover what recordings cannot.

const layer = (handler: Parameters<typeof dynamicResponse>[0]) =>
  SpeechClient.layer.pipe(Layer.provideMerge(dynamicResponse(handler)))

const respond = (body: string | ReadableStream<Uint8Array>, contentType: string) =>
  layer((input) => Effect.succeed(input.respond(body, { headers: { "content-type": contentType } })))

const collect = <E, R>(stream: Stream.Stream<SpeechEvent, E, R>) =>
  stream.pipe(
    Stream.runCollect,
    Effect.map((events) => Array.from(events)),
  )

const openai = OpenAI.configure({ apiKey: "test", baseURL: "https://openai.test/v1" }).speech("gpt-4o-mini-tts")
const elevenlabs = ElevenLabs.configure({ apiKey: "test", baseURL: "https://elevenlabs.test" }).speech(
  "eleven_flash_v2_5",
)
const cartesia = Cartesia.configure({ apiKey: "test", baseURL: "https://cartesia.test" }).speech("sonic-3")
const google = Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).speech(
  "gemini-2.5-flash-preview-tts",
)
const google38 = Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).speech(
  "gemini-3.8-flash-tts",
)
const google38Lite = Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).speech(
  "gemini-3.8-flash-lite-tts",
)
const deepgram = Deepgram.configure({ apiKey: "test", baseURL: "https://deepgram.test" }).speech("aura-2-thalia-en")
const voice = "JBFqnCBsd6RMkjVDRZzb"

describe("Speech", () => {
  it.effect("preserves Google's WAV output instead of describing it as raw PCM", () =>
    Effect.gen(function* () {
      const bytes = new TextEncoder().encode("RIFF....WAVEfmt ")
      const response = yield* Speech.generate({ model: google38, text: "Hi" }).pipe(
        Effect.provide(
          respond(
            JSON.stringify({
              candidates: [
                {
                  content: { parts: [{ inlineData: { mimeType: "audio/wav", data: Base64.encode(bytes) } }] },
                  finishReason: "STOP",
                },
              ],
            }),
            "application/json",
          ),
        ),
      )
      expect(response.audio.mediaType).toBe("audio/wav")
      expect(response.audio.info?.format).toBe("wav")
      expect(response.audio.info?.encoding).toBeUndefined()
      expect(yield* response.audio.bytes()).toEqual(bytes)
    }),
  )

  it.effect("describes OpenAI audio in the format the request body actually asked for", () =>
    Effect.gen(function* () {
      const [pcm, wav] = yield* Effect.all([
        Speech.generate({ model: openai, text: "Hi", format: "mp3", providerOptions: { response_format: "pcm" } }),
        Speech.generate({ model: openai, text: "Hi", http: { body: { response_format: "wav" } } }),
      ]).pipe(Effect.provide(respond("\u0001\u0002", "application/octet-stream")))
      expect(pcm.audio.mediaType).toBe("audio/pcm")
      expect(pcm.audio.info).toEqual({ format: "pcm", encoding: "pcm_s16le", sampleRate: 24000, channels: 1 })
      expect(wav.audio.mediaType).toBe("audio/wav")
      expect(wav.audio.info?.format).toBe("wav")
    }),
  )

  it.effect("describes Deepgram raw encodings in their default WAV container", () =>
    Effect.gen(function* () {
      const response = yield* Speech.generate({ model: deepgram, text: "Hi", providerOptions: { encoding: "mulaw" } })
      expect(response.audio.mediaType).toBe("audio/wav")
      expect(response.audio.info?.format).toBe("wav")
    }).pipe(Effect.provide(respond("RIFF....WAVEfmt ", "audio/wav"))),
  )

  it.effect("always gives headerless Deepgram PCM a sample rate", () =>
    Effect.gen(function* () {
      const [requested, defaulted] = yield* Effect.all([
        Speech.generate({
          model: deepgram,
          text: "Hi",
          providerOptions: { encoding: "mulaw", container: "none", sampleRate: 16000 },
        }),
        Speech.generate({ model: deepgram, text: "Hi", providerOptions: { encoding: "alaw", container: "none" } }),
      ]).pipe(Effect.provide(respond("\u0001\u0002", "audio/basic")))
      expect(requested.audio.info).toEqual({ format: "pcm", encoding: "pcm_mulaw", sampleRate: 16000, channels: 1 })
      expect(defaulted.audio.info).toEqual({ format: "pcm", encoding: "pcm_alaw", sampleRate: 8000, channels: 1 })
    }),
  )

  it.effect("rejects raw PCM for Gemini 3.8 unary requests before sending", () =>
    Effect.gen(function* () {
      const errors = yield* Effect.all(
        [google38, google38Lite].map((model) =>
          Speech.generate({ model, text: "Hi", format: "pcm" }).pipe(Effect.flip),
        ),
      ).pipe(Effect.provide(layer(() => Effect.die("An unsupported request reached the network"))))
      expect(errors.map((error) => error.reason._tag)).toEqual(["UnsupportedOperation", "UnsupportedOperation"])
    }),
  )

  it.effect("rejects what a provider cannot produce before sending anything", () =>
    Effect.gen(function* () {
      const errors = yield* Effect.all(
        [
          Speech.generate({ model: openai, text: "Hi", timestamps: true }),
          Speech.generate({ model: openai, text: "Hi", format: "ogg" }),
          Speech.generate({ model: google, text: "Hi", format: "mp3" }),
          Speech.generate({ model: google, text: "Hi", instructions: "Warm." }),
          collect(Speech.stream({ model: elevenlabs, text: "Hi", voice, format: "wav" })),
          collect(Speech.stream({ model: cartesia, text: "Hi", voice, format: "mp3" })),
          Speech.generate({ model: deepgram, text: "Hi", voice: "thalia" }),
        ].map((effect) => Effect.flip(effect)),
      )
      expect(errors.map((error) => [error.reason._tag, "operation" in error.reason && error.reason.operation])).toEqual(
        [
          ["UnsupportedOperation", "media.timestamps"],
          ["UnsupportedOperation", "media.format"],
          ["UnsupportedOperation", "media.format"],
          ["UnsupportedOperation", "media.instructions"],
          ["UnsupportedOperation", "media.format"],
          ["UnsupportedOperation", "media.format"],
          ["UnsupportedOperation", "media.voice"],
        ],
      )
      expect(errors[1].reason).toMatchObject({ provider: "openai", route: "openai-speech" })
      expect(errors[2].reason).toMatchObject({ provider: "google", route: "google-speech" })
    }).pipe(Effect.provide(layer(() => Effect.die("an unsupported request reached the network")))),
  )

  it.effect("treats timestamps: false as not asking for timestamps on routes that cannot return them", () =>
    Effect.gen(function* () {
      const bytes = Uint8Array.from([1, 2, 3])
      const gemini = JSON.stringify({
        candidates: [
          {
            content: { parts: [{ inlineData: { mimeType: "audio/L16;codec=pcm;rate=24000", data: "AQID" } }] },
            finishReason: "STOP",
          },
        ],
      })
      const responses = yield* Effect.all([
        Speech.generate({ model: openai, text: "Hi", timestamps: false }).pipe(
          Effect.provide(respond(new Blob([bytes]).stream(), "audio/mpeg")),
        ),
        Speech.generate({ model: google, text: "Hi", timestamps: false }).pipe(
          Effect.provide(respond(gemini, "application/json")),
        ),
        Speech.generate({ model: deepgram, text: "Hi", timestamps: false }).pipe(
          Effect.provide(respond(new Blob([bytes]).stream(), "audio/mpeg")),
        ),
      ])
      for (const response of responses) expect(yield* response.audio.bytes()).toEqual(bytes)

      const errors = yield* Effect.all(
        [openai, google, deepgram].map((model) =>
          Speech.generate({ model, text: "Hi", timestamps: true }).pipe(Effect.flip),
        ),
      ).pipe(Effect.provide(layer(() => Effect.die("an unsupported request reached the network"))))
      expect(errors.map((error) => [error.reason._tag, "operation" in error.reason && error.reason.operation])).toEqual(
        [
          ["UnsupportedOperation", "media.timestamps"],
          ["UnsupportedOperation", "media.timestamps"],
          ["UnsupportedOperation", "media.timestamps"],
        ],
      )
    }),
  )

  it.effect("classifies stream failures and keeps the provider payload and HTTP context", () =>
    Effect.gen(function* () {
      const badFrame = JSON.stringify({ type: "speech.audio.delta", audio: "not base64!" })
      const cartesiaError = { type: "error", done: true, status_code: 400, title: "Invalid model", message: "Nope" }
      const blocked = JSON.stringify({ promptFeedback: { blockReason: "PROHIBITED_CONTENT" } })
      const [truncated, invalid, provider, policy] = yield* Effect.all(
        [
          collect(Speech.stream({ model: openai, text: "Hi" })).pipe(
            Effect.provide(respond(sseEvents({ type: "speech.audio.delta", audio: "AQI=" }), "text/event-stream")),
          ),
          collect(Speech.stream({ model: openai, text: "Hi" })).pipe(
            Effect.provide(respond(sseEvents(badFrame), "text/event-stream")),
          ),
          collect(Speech.stream({ model: cartesia, text: "Hi", voice })).pipe(
            Effect.provide(respond(sseEvents(cartesiaError), "text/event-stream")),
          ),
          Speech.generate({ model: google, text: "Hi" }).pipe(Effect.provide(respond(blocked, "application/json"))),
        ].map((effect) => Effect.flip(effect)),
      )

      expect(truncated.reason).toMatchObject({ _tag: "InvalidProviderOutput", classification: "incomplete-stream" })
      expect(truncated.reason.http?.status).toBe(200)
      expect(invalid.reason).toMatchObject({ _tag: "InvalidProviderOutput", body: badFrame })
      expect(invalid.reason.http?.status).toBe(200)
      expect(provider.reason).toMatchObject({ _tag: "InvalidRequest", body: JSON.stringify(cartesiaError) })
      expect(provider.message).toBe("Cartesia stream failed (Invalid model): Nope")
      expect(policy.reason).toMatchObject({ _tag: "ContentPolicy", body: blocked })
    }),
  )

  it.effect("ignores unknown OpenAI speech stream events", () =>
    Effect.gen(function* () {
      const events = yield* collect(Speech.stream({ model: openai, text: "Hi" })).pipe(
        Effect.provide(
          respond(
            sseEvents(
              { type: "speech.audio.delta", audio: "AQI=" },
              { type: "speech.audio.future", value: 1 },
              { type: "speech.audio.done" },
            ),
            "text/event-stream",
          ),
        ),
      )
      expect(events.map((event) => event.type)).toEqual(["audio-delta", "finish"])
    }),
  )

  it.effect("surfaces Gemini speech that ended without STOP instead of returning it as complete", () =>
    Effect.gen(function* () {
      const document = (finishReason?: string) =>
        JSON.stringify({
          candidates: [
            {
              content: { parts: [{ inlineData: { mimeType: "audio/L16;codec=pcm;rate=24000", data: "AQI=" } }] },
              finishReason,
            },
          ],
        })
      const withheld = JSON.stringify({ candidates: [{ finishReason: "SAFETY" }] })
      const generate = (body: string) =>
        Speech.generate({ model: google, text: "Hi" }).pipe(Effect.provide(respond(body, "application/json")))

      const truncated = yield* generate(document()).pipe(Effect.flip)
      const partial = yield* generate(document("MAX_TOKENS"))
      const policy = yield* generate(withheld).pipe(Effect.flip)

      expect(truncated.reason).toMatchObject({ _tag: "InvalidProviderOutput", classification: "incomplete-stream" })
      expect(yield* partial.audio.bytes()).toEqual(Uint8Array.from([1, 2]))
      expect(partial.providerMetadata?.google).toMatchObject({ finishReason: "MAX_TOKENS" })
      expect(policy.reason).toMatchObject({ _tag: "ContentPolicy", body: withheld })
    }),
  )

  it.effect("parses ElevenLabs timestamped records split across network chunks", () =>
    Effect.gen(function* () {
      const record = (bytes: ReadonlyArray<number>, character: string, start: number) =>
        JSON.stringify({
          audio_base64: Base64.encode(Uint8Array.from(bytes)),
          alignment: {
            characters: [character],
            character_start_times_seconds: [start],
            character_end_times_seconds: [start + 0.1],
          },
        })
      const first = record([1, 2], "H", 0)
      const second = record([3], "i", 0.1)
      const encoder = new TextEncoder()
      const body = new ReadableStream<Uint8Array>({
        start(controller) {
          controller.enqueue(encoder.encode(`${first}\n${second.slice(0, 10)}`))
          controller.enqueue(encoder.encode(`${second.slice(10)}\n`))
          controller.close()
        },
      })

      const events = yield* collect(Speech.stream({ model: elevenlabs, text: "Hi", voice, timestamps: true })).pipe(
        Effect.provide(respond(body, "application/json")),
      )

      expect(events.map((event) => event.type)).toEqual([
        "audio-delta",
        "timestamps",
        "audio-delta",
        "timestamps",
        "finish",
      ])
      expect(events.filter(SpeechEvent.is.timestamps).flatMap((event) => event.items)).toEqual([
        { text: "H", startSeconds: 0, endSeconds: 0.1 },
        { text: "i", startSeconds: 0.1, endSeconds: 0.2 },
      ])
      const finish = events.find(SpeechEvent.is.finish)
      expect(yield* finish!.audio.bytes()).toEqual(Uint8Array.from([1, 2, 3]))
    }),
  )
})