defmodule BullX.LLM.Providers.Google do
  @moduledoc """
  Google Gemini provider – built on the OpenAI baseline defaults with Gemini-specific customizations.

  ## Implementation

  Uses built-in defaults with custom encoding/decoding to translate between OpenAI format and Gemini API format.

  ## Google-Specific Extensions

  Beyond standard OpenAI parameters, Google supports:
  - `google_api_version` - Select API version ("v1" or "v1beta"). Defaults to "v1" for production stability.
    Set to "v1beta" to enable beta features like Google Search grounding.
  - `google_safety_settings` - List of safety filter configurations
  - `google_candidate_count` - Number of response candidates to generate (default: 1)
  - `google_grounding` - Enable Google Search grounding (built-in web search). Requires `google_api_version: "v1beta"`
  - `google_thinking_budget` - Thinking token budget for Gemini 2.5 models (cannot be combined with `google_thinking_level`)
  - `google_thinking_level` - Thinking level for Gemini 3+ models (`:minimal`, `:low`, `:medium`, `:high`). Cannot be combined with `google_thinking_budget`
  - `cached_content` - Reference to cached content for 90% cost savings (see Context Caching below)
  - `dimensions` - Number of dimensions for embedding vectors
  - `task_type` - Task type for embeddings (e.g., RETRIEVAL_QUERY)
  - `response_modalities` - Control output modalities for image generation (e.g., ["IMAGE"] for image-only)

  See `provider_schema/0` for the complete Google-specific schema and
  `ReqLLM.Provider.Options` for inherited OpenAI parameters.

  ## Context Caching

  Gemini models support explicit context caching to reduce costs by up to 90% when reusing large amounts of content:

      # Create a cache with large context
      {:ok, cache} = BullX.LLM.Providers.Google.CachedContent.create(
        provider: :google,
        model: "gemini-2.5-flash",
        api_key: System.get_env("GOOGLE_API_KEY"),
        contents: [%{role: "user", parts: [%{text: large_document}]}],
        system_instruction: "You are a helpful assistant.",
        ttl: "3600s"
      )

      # Use the cache in requests (90% discount on cached tokens!)
      {:ok, response} = ReqLLM.generate_text(
        "google:gemini-2.5-flash",
        "Question about the document?",
        provider_options: [cached_content: cache.name]
      )

      # Check token usage - note the cached_tokens field
      IO.inspect(response.usage)
      # %{input_tokens: 50, cached_tokens: 10000, output_tokens: 100, ...}

  See `BullX.LLM.Providers.Google.CachedContent` for full API documentation.

  ## API Version Selection

  The provider defaults to Google's v1beta API which supports all features including function calling
  (tools) and Google Search grounding. For legacy compatibility, you can force v1 by setting
  `google_api_version: "v1"`, but note that v1 does not support function calling or grounding:

      ReqLLM.generate_text(
        "google:gemini-2.5-flash",
        "What are today's tech headlines?",
        provider_options: [
          google_grounding: %{enable: true}
        ]
      )

  **Note**: Setting `google_api_version: "v1"` with function calling (tools) or grounding will return an error.

  ## Configuration

      # Add to .env file (automatically loaded)
      GOOGLE_API_KEY=AIza...
  """

  use ReqLLM.Provider,
    id: :google,
    default_base_url: "https://generativelanguage.googleapis.com/v1beta",
    default_env_key: "GOOGLE_API_KEY"

  import ReqLLM.Provider.Utils,
    only: [maybe_put: 3, ensure_parsed_body: 1, sanitize_url: 1]

  require Logger

  @provider_schema [
    google_api_version: [
      type: {:in, ["v1", "v1beta"]},
      doc:
        "Google API version. Default is 'v1beta'. Set to 'v1' only if you need legacy API behavior. Note: function calling (tools) and grounding require 'v1beta'."
    ],
    google_safety_settings: [
      type: {:list, :map},
      doc: "Safety filter settings for content generation"
    ],
    google_candidate_count: [
      type: :pos_integer,
      default: 1,
      doc: "Number of response candidates to generate"
    ],
    google_thinking_budget: [
      type: :non_neg_integer,
      doc:
        "Thinking token budget for Gemini 2.5 models (0 disables thinking, omit for dynamic). Cannot be combined with google_thinking_level."
    ],
    google_thinking_level: [
      type: {:or, [:atom, :string]},
      doc:
        "Thinking level for Gemini 3+ models (e.g. :low, :medium, :high, or \"low\", \"medium\", \"high\"). Passed directly to the Gemini API. Cannot be combined with google_thinking_budget."
    ],
    google_grounding: [
      type: :map,
      doc:
        "Enable Google Search grounding - allows model to search the web. Set to %{enable: true} for modern models, or %{dynamic_retrieval: %{mode: \"MODE_DYNAMIC\", dynamic_threshold: 0.7}} for Gemini 1.5 legacy support. Requires v1beta (default)."
    ],
    google_url_context: [
      type: {:or, [:boolean, :map]},
      doc:
        "Enable URL context grounding - allows model to fetch and use content from specific URLs. Pass `true` or a map with options. Requires v1beta (default)."
    ],
    dimensions: [
      type: :pos_integer,
      doc:
        "Number of dimensions for the embedding vector (128-3072, recommended: 768, 1536, or 3072)"
    ],
    task_type: [
      type: :string,
      doc:
        "Task type for embedding (e.g., RETRIEVAL_QUERY, RETRIEVAL_DOCUMENT, SEMANTIC_SIMILARITY)"
    ],
    cached_content: [
      type: :string,
      doc:
        "Reference to a previously created cached content. Use the cache name/ID returned from CachedContent creation API."
    ],
    google_auth_header: [
      type: :boolean,
      default: false,
      doc:
        "Use x-goog-api-key header for authentication instead of URL query parameter. Required for OpenAI-compatible API proxies."
    ],
    response_modalities: [
      type: {:list, {:in, ["TEXT", "IMAGE"]}},
      doc:
        "Control output modalities for image generation. List of \"TEXT\" and/or \"IMAGE\". Default is [\"TEXT\", \"IMAGE\"]. Use [\"IMAGE\"] for image-only responses."
    ]
  ]

  defp has_grounding?(opts) do
    provider = Keyword.get(opts, :provider_options, [])

    case Keyword.get(provider, :google_grounding) do
      m when is_map(m) and map_size(m) > 0 -> true
      _ -> false
    end
  end

  defp has_tools?(opts) do
    case Keyword.get(opts, :tools) do
      tools when is_list(tools) and tools != [] -> true
      _ -> false
    end
  end

  defp resolve_api_version(opts) when is_list(opts) do
    provider = Keyword.get(opts, :provider_options, [])

    case Keyword.get(provider, :google_api_version) do
      "v1" -> "v1"
      "v1beta" -> "v1beta"
      _ -> nil
    end
  end

  defp effective_base_url(processed_opts) do
    base_url = Keyword.get(processed_opts, :base_url)

    if base_url == default_base_url() or base_url == "https://generativelanguage.googleapis.com" do
      case resolve_api_version(processed_opts) do
        "v1" ->
          "https://generativelanguage.googleapis.com/v1"

        "v1beta" ->
          "https://generativelanguage.googleapis.com/v1beta"

        nil ->
          "https://generativelanguage.googleapis.com/v1beta"
      end
    else
      base_url || "https://generativelanguage.googleapis.com/v1beta"
    end
  end

  @spec list_models(keyword()) :: {:ok, [BullX.LLM.ModelDescriptor.t()]} | {:error, term()}
  def list_models(opts) when is_list(opts) do
    BullX.LLM.ProviderModelDiscovery.list_google(
      Keyword.put(opts, :base_url, model_discovery_base_url(opts))
    )
  end

  defp model_discovery_base_url(opts) do
    case {Keyword.get(opts, :base_url), google_api_version(opts)} do
      {nil, version} ->
        "https://generativelanguage.googleapis.com/#{version}"

      {"https://generativelanguage.googleapis.com", version} ->
        "https://generativelanguage.googleapis.com/#{version}"

      {base_url, _version} ->
        base_url
    end
  end

  defp google_api_version(opts) do
    opts
    |> Keyword.get(:opts, [])
    |> Keyword.get(:provider_options, [])
    |> Keyword.get(:google_api_version, "v1beta")
  end

  defp validate_version_feature_compat(processed_opts) do
    case {resolve_api_version(processed_opts), has_grounding?(processed_opts),
          has_tools?(processed_opts)} do
      {"v1", true, _} ->
        {:error,
         ReqLLM.Error.Invalid.Parameter.exception(
           parameter:
             ~s/google_grounding requires google_api_version: "v1beta" (or remove the v1 override to use the default)/
         )}

      {"v1", _, true} ->
        {:error,
         ReqLLM.Error.Invalid.Parameter.exception(
           parameter:
             ~s/function calling (tools) requires google_api_version: "v1beta" (or remove the v1 override to use the default)/
         )}

      _ ->
        :ok
    end
  end

  @doc """
  Custom prepare_request for chat operations to use Google's specific endpoints.

  Uses Google's :generateContent and :streamGenerateContent endpoints instead
  of the standard OpenAI /chat/completions endpoint.
  """
  @impl ReqLLM.Provider
  def prepare_request(:chat, model_spec, prompt, opts) do
    with {:ok, model} <- ReqLLM.model(model_spec),
         {:ok, context} <- ReqLLM.Context.normalize(prompt, opts),
         opts_with_context = Keyword.put(opts, :context, context),
         {:ok, processed_opts0} <-
           ReqLLM.Provider.Options.process(__MODULE__, :chat, model, opts_with_context),
         :ok <- validate_version_feature_compat(processed_opts0) do
      processed_opts =
        Keyword.put(processed_opts0, :base_url, effective_base_url(processed_opts0))

      http_opts = Keyword.get(processed_opts, :req_http_options, [])

      # Determine endpoint based on streaming
      endpoint =
        if processed_opts[:stream], do: ":streamGenerateContent", else: ":generateContent"

      req_keys =
        __MODULE__.supported_provider_options() ++
          [:context, :operation, :text, :stream, :model, :provider_options, :tools, :tool_choice]

      # Add alt=sse parameter for streaming requests
      base_params = if processed_opts[:stream], do: [alt: "sse"], else: []

      timeout =
        Keyword.get(
          processed_opts,
          :receive_timeout,
          Application.get_env(:req_llm, :receive_timeout, 30_000)
        )

      request =
        Req.new(
          [
            url: "/models/#{model.id}#{endpoint}",
            method: :post,
            params: base_params,
            receive_timeout: timeout
          ] ++ http_opts
        )
        |> Req.Request.register_options(req_keys)
        |> Req.Request.merge_options(
          Keyword.take(processed_opts, req_keys) ++
            [
              model: model.id,
              base_url: processed_opts[:base_url]
            ]
        )
        |> attach(model, processed_opts)

      {:ok, request}
    end
  end

  def prepare_request(:object, model_spec, prompt, opts) do
    if Keyword.has_key?(opts, :tools) and Keyword.get(opts, :tools) != [] do
      {:error,
       ReqLLM.Error.Invalid.Parameter.exception(
         parameter:
           "tools are not supported with :object operation on Google (JSON mode and tool calling are mutually exclusive on Gemini 2.5)"
       )}
    else
      with {:ok, model} <- ReqLLM.model(model_spec),
           {:ok, context} <- ReqLLM.Context.normalize(prompt, opts) do
        opts_with_tokens =
          ReqLLM.Provider.Options.put_model_max_tokens_default(opts, model, fallback: 4096)

        opts_with_tokens =
          case Keyword.get(opts_with_tokens, :max_tokens) do
            tokens when is_integer(tokens) and tokens < 200 ->
              Keyword.put(opts_with_tokens, :max_tokens, 200)

            _tokens ->
              opts_with_tokens
          end

        opts_with_context =
          opts_with_tokens
          |> Keyword.put(:context, context)
          |> Keyword.put(:operation, :object)

        case ReqLLM.Provider.Options.process(__MODULE__, :object, model, opts_with_context) do
          {:ok, processed_opts0} ->
            with :ok <- validate_version_feature_compat(processed_opts0) do
              processed_opts =
                Keyword.put(processed_opts0, :base_url, effective_base_url(processed_opts0))

              http_opts = Keyword.get(processed_opts, :req_http_options, [])

              endpoint =
                if processed_opts[:stream], do: ":streamGenerateContent", else: ":generateContent"

              req_keys =
                __MODULE__.supported_provider_options() ++
                  [
                    :context,
                    :operation,
                    :compiled_schema,
                    :text,
                    :stream,
                    :model,
                    :provider_options,
                    :tools,
                    :tool_choice
                  ]

              base_params = if processed_opts[:stream], do: [alt: "sse"], else: []

              timeout =
                Keyword.get(
                  processed_opts,
                  :receive_timeout,
                  Application.get_env(:req_llm, :receive_timeout, 30_000)
                )

              request =
                Req.new(
                  [
                    url: "/models/#{model.id}#{endpoint}",
                    method: :post,
                    params: base_params,
                    receive_timeout: timeout
                  ] ++ http_opts
                )
                |> Req.Request.register_options(req_keys)
                |> Req.Request.merge_options(
                  Keyword.take(processed_opts, req_keys) ++
                    [
                      model: model.id,
                      base_url: processed_opts[:base_url]
                    ]
                )
                |> attach(model, processed_opts)

              {:ok, request}
            end

          {:error, reason} ->
            {:error, reason}
        end
      end
    end
  end

  def prepare_request(:embedding, model_spec, text, opts) do
    opts_normalized =
      case Keyword.pop(opts, :dimensions) do
        {nil, rest} ->
          rest

        {dimensions_value, rest} ->
          provider_options = Keyword.get(rest, :provider_options, [])
          updated_provider_options = Keyword.put(provider_options, :dimensions, dimensions_value)
          Keyword.put(rest, :provider_options, updated_provider_options)
      end

    with {:ok, model} <- ReqLLM.model(model_spec),
         opts_with_text = Keyword.merge(opts_normalized, text: text, operation: :embedding),
         {:ok, processed_opts0} <-
           ReqLLM.Provider.Options.process(__MODULE__, :embedding, model, opts_with_text),
         :ok <- validate_version_feature_compat(processed_opts0) do
      processed_opts =
        Keyword.put(processed_opts0, :base_url, effective_base_url(processed_opts0))

      http_opts = Keyword.get(processed_opts, :req_http_options, [])

      endpoint =
        if is_list(text),
          do: ":batchEmbedContents",
          else: ":embedContent"

      req_keys =
        __MODULE__.supported_provider_options() ++
          [:context, :operation, :text, :stream, :model, :provider_options]

      timeout =
        Keyword.get(
          processed_opts,
          :receive_timeout,
          Application.get_env(:req_llm, :receive_timeout, 30_000)
        )

      request =
        Req.new(
          [
            url: "/models/#{model.id}#{endpoint}",
            method: :post,
            receive_timeout: timeout
          ] ++ http_opts
        )
        |> Req.Request.register_options(req_keys)
        |> Req.Request.merge_options(
          Keyword.take(processed_opts, req_keys) ++
            [
              model: model.id,
              base_url: processed_opts[:base_url]
            ]
        )
        |> attach(model, processed_opts)

      {:ok, request}
    end
  end

  def prepare_request(:image, model_spec, prompt, opts) do
    with {:ok, model} <- ReqLLM.model(model_spec),
         :ok <- validate_image_n(model, opts),
         {:ok, context} <- image_context(prompt, opts),
         opts_with_context = Keyword.put(opts, :context, context),
         {:ok, processed_opts0} <-
           ReqLLM.Provider.Options.process(__MODULE__, :image, model, opts_with_context),
         :ok <- validate_version_feature_compat(processed_opts0) do
      processed_opts =
        Keyword.put(processed_opts0, :base_url, effective_base_url(processed_opts0))

      processed_opts =
        Keyword.put(processed_opts, :image_n_provided, Keyword.has_key?(opts, :n))

      http_opts = Keyword.get(processed_opts, :req_http_options, [])

      timeout =
        Keyword.get(
          processed_opts,
          :receive_timeout,
          Application.get_env(:req_llm, :image_receive_timeout, 120_000)
        )

      req_keys =
        __MODULE__.supported_provider_options() ++
          [
            :context,
            :operation,
            :model,
            :n,
            :size,
            :aspect_ratio,
            :output_format,
            :response_format,
            :quality,
            :style,
            :seed,
            :negative_prompt,
            :user,
            :provider_options,
            :base_url,
            :image_n_provided
          ]

      request =
        Req.new(
          [
            url: "/models/#{model.id}#{google_image_endpoint(model)}",
            method: :post,
            receive_timeout: timeout
          ] ++ http_opts
        )
        |> Req.Request.register_options(req_keys)
        |> Req.Request.merge_options(
          Keyword.take(processed_opts, req_keys) ++
            [
              operation: :image,
              model: model.id,
              context: context,
              base_url: processed_opts[:base_url]
            ]
        )
        |> attach(model, processed_opts)

      {:ok, request}
    end
  end

  # Delegate all other operations to defaults (which will return appropriate errors)
  def prepare_request(operation, model_spec, input, opts) do
    ReqLLM.Provider.Defaults.prepare_request(__MODULE__, operation, model_spec, input, opts)
  end

  defp image_context(prompt, opts) do
    case Keyword.get(opts, :context) do
      %ReqLLM.Context{} = context -> {:ok, context}
      _ -> ReqLLM.Context.normalize(prompt, opts)
    end
  end

  defp validate_image_n(%LLMDB.Model{} = model, opts) do
    if Keyword.has_key?(opts, :n) and image_n_forbidden?(model) do
      {:error,
       ReqLLM.Error.Invalid.Parameter.exception(
         parameter:
           "n is not supported for gemini-2.5-flash-image or gemini-3-pro-image-preview; specify the image count in the prompt"
       )}
    else
      :ok
    end
  end

  defp image_n_forbidden?(%LLMDB.Model{provider: :google, id: id}) do
    id in ["gemini-2.5-flash-image", "gemini-3-pro-image-preview"]
  end

  defp image_n_forbidden?(_), do: false

  defp google_image_endpoint(%LLMDB.Model{id: id}) do
    if imagen_model_id?(id), do: ":predict", else: ":generateContent"
  end

  @impl ReqLLM.Provider
  def attach(%Req.Request{} = request, model_input, user_opts) do
    %LLMDB.Model{} =
      model =
      case ReqLLM.model(model_input) do
        {:ok, m} -> m
        {:error, err} -> raise err
      end

    if model.provider != provider_id() do
      raise ReqLLM.Error.Invalid.Provider.exception(provider: model.provider)
    end

    api_key = ReqLLM.Keys.get!(model, user_opts)

    # Filter out internal keys before passing to Req
    req_opts = ReqLLM.Provider.Defaults.filter_req_opts(user_opts)

    # Register extra options that might be passed but aren't standard Req options
    extra_option_keys =
      [
        :model,
        :compiled_schema,
        :temperature,
        :max_tokens,
        :app_referer,
        :app_title,
        :fixture,
        :tools,
        :tool_choice,
        :tool_call_id_compat,
        :n,
        :prompt,
        :size,
        :aspect_ratio,
        :output_format,
        :response_format,
        :json_repair,
        :quality,
        :style,
        :negative_prompt,
        :top_p,
        :top_k,
        :frequency_penalty,
        :presence_penalty,
        :seed,
        :stop,
        :user,
        :system_prompt,
        :reasoning_effort,
        :reasoning_token_budget,
        :telemetry_original_opts,
        :stream,
        :provider_options,
        :dimensions,
        :encoding_format
      ] ++
        __MODULE__.supported_provider_options()

    request
    # Google uses query parameter for API key, not Authorization header
    |> Req.Request.register_options(extra_option_keys)
    |> Req.Request.merge_options([model: model.id, params: [key: api_key]] ++ req_opts)
    |> ReqLLM.Step.Error.attach()
    |> ReqLLM.Step.Retry.attach(user_opts)
    |> Req.Request.prepend_request_steps(llm_encode_body: &__MODULE__.encode_body/1)
    |> Req.Request.append_response_steps(llm_decode_response: &__MODULE__.decode_response/1)
    |> ReqLLM.Step.Usage.attach(model)
    |> ReqLLM.Step.Telemetry.attach(model, user_opts)
    |> ReqLLM.Step.Fixture.maybe_attach(model, user_opts)
  end

  @impl ReqLLM.Provider
  def extract_usage(body, model) when is_map(body) do
    case body do
      %{"usageMetadata" => usage_metadata} ->
        usage = normalize_google_usage(usage_metadata)
        tool_usage = google_tool_usage(body, model)
        image_usage = google_image_usage(body)

        usage =
          usage
          |> Map.put(:tool_usage, tool_usage)
          |> maybe_put_image_usage(image_usage)

        {:ok, usage}

      _ ->
        image_usage = google_image_usage(body)

        if map_size(image_usage) > 0 do
          {:ok, %{image_usage: image_usage}}
        else
          {:error, :no_usage_found}
        end
    end
  end

  def extract_usage(_, _), do: {:error, :invalid_body}

  defp normalize_google_usage(usage_metadata) do
    google_usage_from_metadata(usage_metadata)
  end

  defp google_usage_from_metadata(usage_metadata) when is_map(usage_metadata) do
    input =
      google_metadata_count(usage_metadata, "promptTokenCount") ||
        google_token_details_count(usage_metadata["promptTokensDetails"]) ||
        0

    total = google_metadata_count(usage_metadata, "totalTokenCount")
    reasoning = google_metadata_count(usage_metadata, "thoughtsTokenCount") || 0
    cached = google_metadata_count(usage_metadata, "cachedContentTokenCount") || 0
    candidates = google_metadata_count(usage_metadata, "candidatesTokenCount")
    output = google_output_tokens(candidates, reasoning, total, input)

    %{
      input_tokens: input,
      output_tokens: output,
      total_tokens: total || input + output,
      cached_tokens: cached,
      reasoning_tokens: reasoning
    }
  end

  defp google_usage_from_metadata(_usage_metadata), do: google_zero_usage()

  defp google_metadata_count(usage_metadata, key) do
    case Map.get(usage_metadata, key) do
      count when is_integer(count) and count >= 0 -> count
      _ -> nil
    end
  end

  defp google_token_details_count(details) when is_list(details) do
    Enum.reduce(details, nil, fn detail, total ->
      case detail do
        %{"tokenCount" => count} when is_integer(count) and count >= 0 ->
          add_google_token_count(total, count)

        _ ->
          total
      end
    end)
  end

  defp google_token_details_count(_details), do: nil

  defp add_google_token_count(nil, count), do: count
  defp add_google_token_count(total, count), do: total + count

  defp google_output_tokens(candidates, reasoning, _total, _input) when is_integer(candidates) do
    candidates + reasoning
  end

  defp google_output_tokens(_candidates, _reasoning, total, input) when is_integer(total) do
    max(0, total - input)
  end

  defp google_output_tokens(_candidates, reasoning, _total, _input) when reasoning > 0 do
    reasoning
  end

  defp google_output_tokens(_candidates, _reasoning, _total, _input), do: 0

  defp google_zero_usage do
    %{
      input_tokens: 0,
      output_tokens: 0,
      total_tokens: 0,
      cached_tokens: 0,
      reasoning_tokens: 0
    }
  end

  defp google_tool_usage(body, model) do
    queries =
      body
      |> Map.get("candidates", [])
      |> Enum.flat_map(fn candidate ->
        case get_in(candidate, ["groundingMetadata", "webSearchQueries"]) do
          queries when is_list(queries) -> queries
          _ -> []
        end
      end)

    if queries == [] do
      %{}
    else
      unit = ReqLLM.Pricing.tool_unit(model, :web_search)

      count =
        case unit do
          :query -> length(queries)
          "query" -> length(queries)
          _ -> 1
        end

      ReqLLM.Usage.Tool.build(:web_search, count, unit)
    end
  end

  defp google_image_usage(body) when is_map(body) do
    candidates = Map.get(body, "candidates", [])

    count =
      Enum.reduce(candidates, 0, fn candidate, acc ->
        parts = get_in(candidate, ["content", "parts"]) || []
        acc + ReqLLM.Usage.Image.count_inline_parts(parts)
      end)

    ReqLLM.Usage.Image.build_generated(count)
  end

  defp maybe_put_image_usage(usage, image_usage) do
    if map_size(image_usage) > 0 do
      Map.put(usage, :image_usage, image_usage)
    else
      usage
    end
  end

  def pre_validate_options(_operation, model, opts) do
    {provider_opts, rest} = Keyword.pop(opts, :provider_options, [])

    {effort, provider_opts} = Keyword.pop(provider_opts, :reasoning_effort)
    provider_opts = normalize_response_modalities(provider_opts)

    provider_opts =
      case {effort, gemini_3_or_later?(model)} do
        {nil, _} ->
          provider_opts

        {effort_value, true} ->
          case Keyword.fetch(provider_opts, :google_thinking_level) do
            {:ok, _existing} ->
              provider_opts

            :error ->
              level = translate_reasoning_effort_to_level(effort_value)
              Keyword.put(provider_opts, :google_thinking_level, level)
          end

        {effort_value, false} ->
          budget = translate_reasoning_effort_to_budget(effort_value, model)

          case Keyword.fetch(provider_opts, :google_thinking_budget) do
            {:ok, existing} when is_integer(existing) and existing > 0 ->
              provider_opts

            {:ok, 0} ->
              Keyword.put(provider_opts, :google_thinking_budget, budget)

            :error ->
              Keyword.put(provider_opts, :google_thinking_budget, budget)
          end
      end

    {Keyword.put(rest, :provider_options, provider_opts), []}
  end

  defp normalize_response_modalities(provider_opts) do
    case Keyword.fetch(provider_opts, :response_modalities) do
      {:ok, modalities} when is_list(modalities) ->
        normalized = Enum.map(modalities, &normalize_response_modality/1)
        Keyword.put(provider_opts, :response_modalities, normalized)

      _ ->
        provider_opts
    end
  end

  defp normalize_response_modality(modality) when is_atom(modality) do
    modality
    |> Atom.to_string()
    |> normalize_response_modality()
  end

  defp normalize_response_modality(modality) when is_binary(modality) do
    modality
    |> String.trim()
    |> String.upcase()
  end

  defp normalize_response_modality(modality), do: modality

  defp translate_reasoning_effort_to_budget(:none, _model), do: 0
  defp translate_reasoning_effort_to_budget(:minimal, _model), do: 2_048
  defp translate_reasoning_effort_to_budget(:low, _model), do: 4_096
  defp translate_reasoning_effort_to_budget(:medium, _model), do: 8_192
  defp translate_reasoning_effort_to_budget(:high, _model), do: 16_384
  defp translate_reasoning_effort_to_budget(:xhigh, _model), do: 32_768

  defp translate_reasoning_effort_to_budget("none", model),
    do: translate_reasoning_effort_to_budget(:none, model)

  defp translate_reasoning_effort_to_budget("minimal", model),
    do: translate_reasoning_effort_to_budget(:minimal, model)

  defp translate_reasoning_effort_to_budget("low", model),
    do: translate_reasoning_effort_to_budget(:low, model)

  defp translate_reasoning_effort_to_budget("medium", model),
    do: translate_reasoning_effort_to_budget(:medium, model)

  defp translate_reasoning_effort_to_budget("high", model),
    do: translate_reasoning_effort_to_budget(:high, model)

  defp translate_reasoning_effort_to_budget("xhigh", model),
    do: translate_reasoning_effort_to_budget(:xhigh, model)

  defp translate_reasoning_effort_to_budget(budget, _model) when is_integer(budget), do: budget
  defp translate_reasoning_effort_to_budget(_unknown, _model), do: 8_192

  defp translate_reasoning_effort_to_level(:none), do: :minimal
  defp translate_reasoning_effort_to_level(:minimal), do: :minimal
  defp translate_reasoning_effort_to_level(:low), do: :low
  defp translate_reasoning_effort_to_level(:medium), do: :medium
  defp translate_reasoning_effort_to_level(:high), do: :high
  defp translate_reasoning_effort_to_level(:xhigh), do: :high

  defp translate_reasoning_effort_to_level("none"), do: :minimal
  defp translate_reasoning_effort_to_level("minimal"), do: :minimal
  defp translate_reasoning_effort_to_level("low"), do: :low
  defp translate_reasoning_effort_to_level("medium"), do: :medium
  defp translate_reasoning_effort_to_level("high"), do: :high
  defp translate_reasoning_effort_to_level("xhigh"), do: :high
  defp translate_reasoning_effort_to_level(_unknown), do: :medium

  @impl ReqLLM.Provider
  def translate_options(:image, _model, opts) do
    opts =
      case {Keyword.get(opts, :aspect_ratio), Keyword.get(opts, :size)} do
        {ratio, _} when is_binary(ratio) and ratio != "" ->
          opts

        {nil, {w, h}} when is_integer(w) and is_integer(h) ->
          Keyword.put(opts, :aspect_ratio, infer_aspect_ratio(w, h))

        {nil, size_str} when is_binary(size_str) ->
          case parse_size(size_str) do
            {:ok, {w, h}} -> Keyword.put(opts, :aspect_ratio, infer_aspect_ratio(w, h))
            :error -> opts
          end

        _ ->
          opts
      end

    {opts, []}
  end

  def translate_options(_operation, model, opts) do
    {reasoning_budget, opts} = Keyword.pop(opts, :reasoning_token_budget)
    {reasoning_effort, opts} = Keyword.pop(opts, :reasoning_effort)

    opts =
      cond do
        reasoning_budget ->
          Keyword.put(opts, :google_thinking_budget, reasoning_budget)

        reasoning_effort && gemini_3_or_later?(model) ->
          level = translate_reasoning_effort_to_level(reasoning_effort)
          Keyword.put(opts, :google_thinking_level, level)

        reasoning_effort ->
          budget = translate_reasoning_effort_to_budget(reasoning_effort, nil)
          Keyword.put(opts, :google_thinking_budget, budget)

        true ->
          opts
      end

    case Keyword.pop(opts, :stream?) do
      {nil, rest} ->
        {rest, []}

      {stream_value, rest} ->
        {Keyword.put(rest, :stream, stream_value), []}
    end
  end

  # Req pipeline steps
  @impl ReqLLM.Provider
  def encode_body(request) do
    body =
      case request.options[:operation] do
        :embedding ->
          encode_embedding_body(request)

        :image ->
          encode_image_body(request)

        :object ->
          encode_object_body(request)

        _ ->
          encode_chat_body(request)
      end

    request
    |> put_in([Access.key!(:options), :json], body)
  end

  defp encode_image_body(request) do
    if imagen_model_id?(request.options[:model]) do
      encode_imagen_image_body(request)
    else
      encode_gemini_image_body(request)
    end
  end

  defp encode_gemini_image_body(request) do
    model_name = request.options[:model]

    {system_instruction, contents} =
      case request.options[:context] do
        %ReqLLM.Context{} = ctx ->
          encoded =
            ctx
            |> normalize_context_video_urls()
            |> ReqLLM.Provider.Defaults.encode_context_to_openai_format(model_name)

          messages = encoded[:messages] || []
          split_messages_for_gemini(messages, model_name)

        _ ->
          split_messages_for_gemini(request.options[:messages] || [], model_name)
      end

    # Note: We intentionally keep the role field in contents.
    # Experiments show that including "role": "user" improves multi-image
    # generation success rate (70% vs 50% for well-phrased prompts).

    generation_config =
      %{}
      |> maybe_put_google_aspect_ratio(request.options[:aspect_ratio])
      |> maybe_put_google_response_modalities(request.options[:response_modalities])
      |> maybe_put(:candidateCount, image_candidate_count(request.options))

    generation_config = if generation_config != %{}, do: generation_config

    %{}
    |> maybe_put(:systemInstruction, system_instruction)
    |> Map.put(:contents, contents)
    |> maybe_put(:generationConfig, generation_config)
  end

  defp encode_imagen_image_body(request) do
    prompt = imagen_prompt(request.options[:context])

    if prompt == "" do
      raise ReqLLM.Error.Invalid.Parameter.exception(
              parameter: "Google Imagen models require a text prompt"
            )
    end

    parameters =
      %{}
      |> maybe_put(:sampleCount, image_candidate_count(request.options))
      |> maybe_put(:aspectRatio, request.options[:aspect_ratio])
      |> maybe_put(:sampleImageSize, imagen_sample_image_size(request.options[:size]))
      |> maybe_put(:outputOptions, imagen_output_options(request.options[:output_format]))

    %{}
    |> Map.put(:instances, [%{prompt: prompt}])
    |> maybe_put(:parameters, if(parameters == %{}, do: nil, else: parameters))
  end

  defp image_candidate_count(opts) when is_list(opts) do
    if Keyword.get(opts, :image_n_provided, false) do
      case Keyword.fetch(opts, :n) do
        {:ok, value} -> value
        :error -> nil
      end
    end
  end

  defp image_candidate_count(opts) when is_map(opts) do
    if Map.get(opts, :image_n_provided, false) and Map.has_key?(opts, :n) do
      Map.get(opts, :n)
    end
  end

  defp image_candidate_count(_), do: nil

  defp imagen_prompt(%ReqLLM.Context{messages: messages}) do
    messages
    |> Enum.map(&imagen_message_prompt/1)
    |> Enum.reject(&(&1 in [nil, ""]))
    |> Enum.join("\n\n")
    |> String.trim()
  end

  defp imagen_prompt(_), do: ""

  defp imagen_message_prompt(%ReqLLM.Message{role: role, content: content}) do
    prompt =
      content
      |> List.wrap()
      |> Enum.map(&imagen_content_prompt/1)
      |> Enum.reject(&(&1 in [nil, ""]))
      |> Enum.join("\n")
      |> String.trim()

    case {role, prompt} do
      {_, ""} -> nil
      {:user, prompt} -> prompt
      {role, prompt} -> "#{role}: #{prompt}"
    end
  end

  defp imagen_message_prompt(_), do: nil

  defp imagen_content_prompt(%ReqLLM.Message.ContentPart{type: :text, text: text})
       when is_binary(text),
       do: text

  defp imagen_content_prompt(%{type: :text, text: text}) when is_binary(text), do: text
  defp imagen_content_prompt(text) when is_binary(text), do: text
  defp imagen_content_prompt(_), do: nil

  defp imagen_output_options(nil), do: nil

  defp imagen_output_options(output_format) do
    %{mimeType: google_image_mime_type(output_format)}
  end

  defp imagen_sample_image_size({w, h}) when is_integer(w) and is_integer(h) do
    imagen_sample_image_size("#{w}x#{h}")
  end

  defp imagen_sample_image_size(size) when is_binary(size) do
    case String.downcase(size) do
      "1024x1024" -> "1K"
      "2048x2048" -> "2K"
      _ -> nil
    end
  end

  defp imagen_sample_image_size(_), do: nil

  defp google_image_mime_type(:png), do: "image/png"
  defp google_image_mime_type(:jpeg), do: "image/jpeg"
  defp google_image_mime_type(:webp), do: "image/webp"
  defp google_image_mime_type(format) when is_binary(format), do: format
  defp google_image_mime_type(_), do: "image/png"

  defp maybe_put_google_aspect_ratio(config, nil), do: config

  defp maybe_put_google_aspect_ratio(config, ratio) when is_binary(ratio) do
    Map.put(
      config,
      "imageConfig",
      Map.put(Map.get(config, "imageConfig", %{}), "aspectRatio", ratio)
    )
  end

  defp maybe_put_google_aspect_ratio(config, _), do: config

  defp maybe_put_google_response_modalities(config, nil), do: config

  defp maybe_put_google_response_modalities(config, modalities) when is_list(modalities) do
    Map.put(config, "responseModalities", modalities)
  end

  defp maybe_put_google_response_modalities(config, _), do: config

  defp parse_size(size) when is_binary(size) do
    case String.split(size, "x") do
      [w, h] ->
        with {w_i, ""} <- Integer.parse(w),
             {h_i, ""} <- Integer.parse(h),
             true <- w_i > 0 and h_i > 0 do
          {:ok, {w_i, h_i}}
        else
          _ -> :error
        end

      _ ->
        :error
    end
  end

  defp infer_aspect_ratio(w, h) when is_integer(w) and is_integer(h) and w > 0 and h > 0 do
    gcd = Integer.gcd(w, h)
    "#{div(w, gcd)}:#{div(h, gcd)}"
  end

  defp encode_chat_body(request) do
    model_name = request.options[:model]

    {system_instruction, contents} =
      case request.options[:context] do
        %ReqLLM.Context{} = ctx ->
          # Convert OpenAI-style context to Gemini format
          encoded =
            ctx
            |> normalize_context_video_urls()
            |> ReqLLM.Provider.Defaults.encode_context_to_openai_format(model_name)

          messages = encoded[:messages] || []
          split_messages_for_gemini(messages, model_name)

        _ ->
          split_messages_for_gemini(request.options[:messages] || [], model_name)
      end

    tool_config = build_google_tool_config(request.options[:tool_choice])

    grounding_tools = build_grounding_tools(request.options[:google_grounding])
    url_context_tools = build_url_context_tools(request.options[:google_url_context])
    builtin_tools = grounding_tools ++ url_context_tools

    tools_data =
      case request.options[:tools] do
        tools when is_list(tools) and tools != [] ->
          user_tools = [
            %{functionDeclarations: Enum.map(tools, &ReqLLM.Tool.to_schema(&1, :google))}
          ]

          all_tools = builtin_tools ++ user_tools

          %{tools: all_tools}
          |> maybe_put(:toolConfig, tool_config)

        _ ->
          case builtin_tools do
            [] ->
              %{}

            tools ->
              %{tools: tools}
              |> maybe_put(:toolConfig, tool_config)
          end
      end

    # Build generationConfig with Gemini-specific parameter names
    generation_config =
      %{}
      |> maybe_put(:temperature, request.options[:temperature])
      |> maybe_put(:maxOutputTokens, request.options[:max_tokens])
      |> maybe_put(:topP, request.options[:top_p])
      |> maybe_put(:topK, request.options[:top_k])
      |> maybe_put(:candidateCount, request.options[:google_candidate_count] || 1)
      |> maybe_add_thinking_config(
        request.options[:google_thinking_budget],
        request.options[:google_thinking_level]
      )

    %{}
    |> maybe_put(:cachedContent, request.options[:cached_content])
    |> maybe_put(:systemInstruction, system_instruction)
    |> Map.put(:contents, contents)
    |> Map.merge(tools_data)
    |> maybe_put(:generationConfig, generation_config)
    |> maybe_put(:safetySettings, request.options[:google_safety_settings])
    |> maybe_put(:labels, request.options[:labels])
  end

  defp encode_embedding_body(request) do
    text = request.options[:text]
    model_id = request.options[:id] || request.options[:model]

    build_embedding_body = fn t ->
      %{
        model: "models/#{model_id}",
        content: %{parts: [%{text: t}]}
      }
      |> maybe_put(:outputDimensionality, request.options[:dimensions])
      |> maybe_put(:taskType, request.options[:task_type])
    end

    case text do
      texts when is_list(texts) ->
        requests = Enum.map(texts, build_embedding_body)
        %{requests: requests}

      single_text when is_binary(single_text) ->
        build_embedding_body.(single_text)
    end
  end

  defp encode_object_body(request) do
    model_name = request.options[:model]

    {system_instruction, contents} =
      case request.options[:context] do
        %ReqLLM.Context{} = ctx ->
          encoded =
            ctx
            |> normalize_context_video_urls()
            |> ReqLLM.Provider.Defaults.encode_context_to_openai_format(model_name)

          messages = encoded[:messages] || []
          split_messages_for_gemini(messages, model_name)

        _ ->
          split_messages_for_gemini(request.options[:messages] || [], model_name)
      end

    compiled_schema =
      case request.options do
        opts when is_map(opts) -> Map.get(opts, :compiled_schema)
        opts when is_list(opts) -> Keyword.get(opts, :compiled_schema)
      end

    if !compiled_schema do
      raise ArgumentError, "Missing :compiled_schema in request options for :object operation"
    end

    generation_config =
      %{
        candidateCount: 1,
        responseMimeType: "application/json"
      }
      |> maybe_put(:temperature, request.options[:temperature])
      |> maybe_put(:maxOutputTokens, request.options[:max_tokens])
      |> maybe_put(:topP, request.options[:top_p])
      |> maybe_put(:topK, request.options[:top_k])
      |> maybe_add_thinking_config(
        request.options[:google_thinking_budget],
        request.options[:google_thinking_level]
      )
      |> put_schema_for_model(model_name, compiled_schema)

    %{}
    |> maybe_put(:cachedContent, request.options[:cached_content])
    |> maybe_put(:systemInstruction, system_instruction)
    |> Map.put(:contents, contents)
    |> maybe_put(:generationConfig, generation_config)
    |> maybe_put(:safetySettings, request.options[:google_safety_settings])
    |> maybe_put(:labels, request.options[:labels])
  end

  defp gemini_3_or_later?(%LLMDB.Model{family: family}) when is_binary(family),
    do: String.starts_with?(family, "gemini-3")

  defp gemini_3_or_later?(%LLMDB.Model{id: id}) when is_binary(id),
    do: String.starts_with?(id, "gemini-3")

  defp gemini_3_or_later?(id) when is_binary(id),
    do: String.starts_with?(id, "gemini-3")

  defp gemini_3_or_later?(_), do: false

  defp json_schema_supported?(model_name) when is_binary(model_name) do
    String.starts_with?(model_name, "gemini-2.5-") or model_name == "gemini-2.5" or
      String.starts_with?(model_name, "gemini-3")
  end

  defp json_schema_supported?(_), do: false

  defp put_schema_for_model(generation_config, model_name, compiled_schema) do
    json_schema = ReqLLM.Schema.to_json(compiled_schema.schema)

    if json_schema_supported?(model_name) and json_schema?(json_schema) do
      Map.put(generation_config, :responseJsonSchema, json_schema)
    else
      google_schema = convert_to_google_schema(json_schema)
      Map.put(generation_config, :responseSchema, google_schema)
    end
  end

  defp json_schema?(%{"type" => type}) when is_binary(type) do
    type in ["object", "array", "string", "number", "integer", "boolean", "null"]
  end

  defp json_schema?(_), do: false

  defp convert_to_google_schema(schema) when is_map(schema) do
    schema
    |> Map.delete("additionalProperties")
    |> Map.new(fn {key, value} ->
      case key do
        "type" -> {"type", to_google_type(value)}
        "properties" -> {"properties", convert_properties_to_google(value)}
        "items" when is_map(value) -> {"items", convert_to_google_schema(value)}
        "items" when is_list(value) -> raise_unsupported_schema("tuple arrays not supported")
        other -> {other, value}
      end
    end)
    |> maybe_add_property_ordering()
  end

  defp convert_to_google_schema(value), do: value

  defp to_google_type("object"), do: "OBJECT"
  defp to_google_type("array"), do: "ARRAY"
  defp to_google_type("string"), do: "STRING"
  defp to_google_type("integer"), do: "INTEGER"
  defp to_google_type("number"), do: "NUMBER"
  defp to_google_type("boolean"), do: "BOOLEAN"
  defp to_google_type("null"), do: "NULL"
  defp to_google_type(type), do: type

  defp convert_properties_to_google(properties) when is_map(properties) do
    Map.new(properties, fn {key, value} ->
      {key, convert_to_google_schema(value)}
    end)
  end

  defp maybe_add_property_ordering(schema) when is_map(schema) do
    case Map.get(schema, "properties") do
      properties when is_map(properties) and map_size(properties) > 0 ->
        if Map.has_key?(schema, "propertyOrdering") do
          schema
        else
          ordering = Map.keys(properties)
          Map.put(schema, "propertyOrdering", ordering)
        end

      _ ->
        schema
    end
  end

  defp raise_unsupported_schema(message) do
    raise ReqLLM.Error.Invalid.Parameter, parameter: "schema: #{message}"
  end

  defp normalize_embedding_response(%{"embedding" => %{"values" => values}} = body)
       when is_list(values) do
    %{"data" => [%{"index" => 0, "embedding" => values}]}
    |> maybe_put_embedding_usage_metadata(body)
  end

  defp normalize_embedding_response(%{"embeddings" => embeddings} = body)
       when is_list(embeddings) do
    data =
      embeddings
      |> Enum.with_index()
      |> Enum.map(fn
        {%{"values" => values}, idx} ->
          %{"index" => idx, "embedding" => values}

        {other, idx} ->
          vals = get_in(other, ["embedding", "values"]) || other["values"] || []
          %{"index" => idx, "embedding" => vals}
      end)

    %{"data" => data}
    |> maybe_put_embedding_usage_metadata(body)
  end

  defp normalize_embedding_response(other), do: other

  defp maybe_put_embedding_usage_metadata(normalized, body) do
    case Map.get(body, "usageMetadata") do
      usage_metadata when is_map(usage_metadata) ->
        Map.put(normalized, "usageMetadata", usage_metadata)

      _ ->
        normalized
    end
  end

  @impl ReqLLM.Provider
  def decode_response({req, resp}) do
    case resp.status do
      200 ->
        operation = req.options[:operation]
        is_streaming = req.options[:stream] == true

        case operation do
          :embedding ->
            body = ensure_parsed_body(resp.body)
            normalized = normalize_embedding_response(body)
            {req, %{resp | body: normalized}}

          :image when not is_streaming ->
            model_name = ReqLLM.ModelId.normalize(req.options[:model], "google")
            body = ensure_parsed_body(resp.body)
            merged_response = decode_image_response(req, model_name, body)
            {req, %{resp | body: merged_response}}

          :object when not is_streaming ->
            model_name = ReqLLM.ModelId.normalize(req.options[:model], "google")
            model = LLMDB.Model.new!(%{id: model_name, provider: :google})
            body = ensure_parsed_body(resp.body)

            openai_format = convert_google_json_mode_to_openai_format(body)

            {:ok, response} =
              ReqLLM.Provider.Defaults.decode_response_body_openai_format(openai_format, model)

            response_with_object =
              case ReqLLM.Response.unwrap_object(response, req.options) do
                {:ok, object} -> %{response | object: object}
                {:error, _} -> response
              end

            merged_response =
              ReqLLM.Context.merge_response(
                req.options[:context] || %ReqLLM.Context{messages: []},
                response_with_object
              )

            {req, %{resp | body: merged_response}}

          _ when is_streaming ->
            ReqLLM.Provider.Defaults.default_decode_response({req, resp})

          _ ->
            model_name = ReqLLM.ModelId.normalize(req.options[:model], "google")
            model = LLMDB.Model.new!(%{id: model_name, provider: :google})

            body = ensure_parsed_body(resp.body)

            grounding_metadata = extract_grounding_metadata(body)

            openai_format = convert_google_to_openai_format(body)

            reasoning_details = extract_reasoning_details_from_openai_format(openai_format)

            {:ok, response} =
              ReqLLM.Provider.Defaults.decode_response_body_openai_format(openai_format, model)

            response_with_reasoning = attach_reasoning_details(response, reasoning_details)
            tool_usage = google_tool_usage(body, model)
            image_usage = google_image_usage(body)

            response_with_usage =
              add_usage_details(response_with_reasoning, tool_usage, image_usage)

            response_with_grounding =
              case grounding_metadata do
                nil ->
                  response_with_usage

                grounding_data ->
                  %{
                    response_with_usage
                    | provider_meta:
                        Map.put(
                          response_with_usage.provider_meta,
                          "google",
                          grounding_data
                        )
                  }
              end

            merged_response =
              ReqLLM.Context.merge_response(
                req.options[:context] || %ReqLLM.Context{messages: []},
                response_with_grounding
              )

            {req, %{resp | body: merged_response}}
        end

      status ->
        err =
          ReqLLM.Error.API.Response.exception(
            reason: "Google API error",
            status: status,
            response_body: resp.body
          )

        {req, err}
    end
  end

  defp decode_image_response(req, model_name, %{} = body) do
    if Map.has_key?(body, "predictions") do
      decode_imagen_response(req, model_name, body)
    else
      decode_gemini_image_response(req, model_name, body)
    end
  end

  defp decode_gemini_image_response(req, model_name, %{} = body) do
    parts = extract_candidate_parts(body)

    content_parts =
      parts
      |> Enum.map(&decode_image_part/1)
      |> Enum.reject(&is_nil/1)

    message = %ReqLLM.Message{role: :assistant, content: content_parts}

    usage =
      case Map.get(body, "usageMetadata") do
        usage_metadata when is_map(usage_metadata) -> normalize_google_usage(usage_metadata)
        _ -> %{}
      end

    image_usage = google_image_usage(body)
    usage = maybe_put_image_usage(usage, image_usage)
    usage = if usage != %{}, do: usage

    base_response = %ReqLLM.Response{
      id: image_response_id(),
      model: model_name,
      context: req.options[:context] || %ReqLLM.Context{messages: []},
      message: message,
      object: nil,
      stream?: false,
      stream: nil,
      usage: usage,
      finish_reason: :stop,
      provider_meta: %{"google" => Map.delete(body, "candidates")},
      error: nil
    }

    ReqLLM.Context.merge_response(base_response.context, base_response)
  end

  defp decode_imagen_response(req, model_name, %{} = body) do
    content_parts =
      body
      |> Map.get("predictions", [])
      |> Enum.map(&decode_imagen_prediction/1)
      |> Enum.reject(&is_nil/1)

    base_response = %ReqLLM.Response{
      id: image_response_id(),
      model: model_name,
      context: req.options[:context] || %ReqLLM.Context{messages: []},
      message: %ReqLLM.Message{role: :assistant, content: content_parts},
      object: nil,
      stream?: false,
      stream: nil,
      usage: nil,
      finish_reason: :stop,
      provider_meta: %{"google" => Map.delete(body, "predictions")},
      error: nil
    }

    ReqLLM.Context.merge_response(base_response.context, base_response)
  end

  defp extract_candidate_parts(%{"candidates" => candidates}) when is_list(candidates) do
    Enum.flat_map(candidates, fn
      %{"content" => %{"parts" => parts}} when is_list(parts) -> parts
      _ -> []
    end)
  end

  defp extract_candidate_parts(_), do: []

  defp decode_image_part(%{"text" => text}) when is_binary(text) and text != "" do
    %ReqLLM.Message.ContentPart{type: :text, text: text}
  end

  defp decode_image_part(%{"inlineData" => inline}) when is_map(inline) do
    decode_inline_data(inline)
  end

  defp decode_image_part(%{"inline_data" => inline}) when is_map(inline) do
    decode_inline_data(inline)
  end

  defp decode_image_part(_), do: nil

  defp decode_imagen_prediction(%{"bytesBase64Encoded" => b64, "mimeType" => mime_type})
       when is_binary(b64) and is_binary(mime_type) do
    %ReqLLM.Message.ContentPart{type: :image, data: Base.decode64!(b64), media_type: mime_type}
  end

  defp decode_imagen_prediction(%{"gcsUri" => uri, "mimeType" => mime_type})
       when is_binary(uri) and is_binary(mime_type) do
    ReqLLM.Message.ContentPart.image_url(uri, %{media_type: mime_type})
  end

  defp decode_imagen_prediction(_), do: nil

  defp decode_inline_data(%{"data" => b64, "mimeType" => mime_type})
       when is_binary(b64) and is_binary(mime_type) do
    %ReqLLM.Message.ContentPart{type: :image, data: Base.decode64!(b64), media_type: mime_type}
  end

  defp decode_inline_data(%{"data" => b64, "mime_type" => mime_type})
       when is_binary(b64) and is_binary(mime_type) do
    %ReqLLM.Message.ContentPart{type: :image, data: Base.decode64!(b64), media_type: mime_type}
  end

  defp decode_inline_data(_), do: nil

  defp image_response_id do
    "img_" <> (:crypto.strong_rand_bytes(12) |> Base.url_encode64(padding: false))
  end

  defp add_usage_details(%ReqLLM.Response{} = response, tool_usage, image_usage) do
    usage = response.usage

    usage =
      if map_size(tool_usage) > 0 do
        Map.put(usage, :tool_usage, tool_usage)
      else
        usage
      end

    usage = maybe_put_image_usage(usage, image_usage)

    if usage == %{} do
      response
    else
      %{response | usage: usage}
    end
  end

  # Helper to build Google toolConfig from OpenAI-style tool_choice
  defp build_google_tool_config(nil), do: nil

  defp build_google_tool_config(%{type: "function", function: %{name: name}}) do
    %{
      functionCallingConfig: %{
        mode: "ANY",
        allowedFunctionNames: [name]
      }
    }
  end

  defp build_google_tool_config(:required), do: build_google_tool_config("required")
  defp build_google_tool_config(:auto), do: build_google_tool_config("auto")
  defp build_google_tool_config(:none), do: build_google_tool_config("none")

  defp build_google_tool_config("required") do
    %{functionCallingConfig: %{mode: "ANY"}}
  end

  defp build_google_tool_config("auto"), do: %{functionCallingConfig: %{mode: "AUTO"}}
  defp build_google_tool_config("none"), do: %{functionCallingConfig: %{mode: "NONE"}}
  defp build_google_tool_config(_), do: nil

  defp build_grounding_tools(nil), do: []
  defp build_grounding_tools(%{enable: true}), do: [%{google_search: %{}}]

  defp build_grounding_tools(%{dynamic_retrieval: config}) when is_map(config) do
    [%{google_search_retrieval: %{dynamic_retrieval_config: config}}]
  end

  defp build_grounding_tools(_), do: []

  defp build_url_context_tools(true), do: [%{url_context: %{}}]
  defp build_url_context_tools(%{} = opts), do: [%{url_context: opts}]
  defp build_url_context_tools(_), do: []

  defp extract_grounding_metadata(%{"candidates" => [candidate | _]}) do
    case candidate do
      %{"groundingMetadata" => metadata} when is_map(metadata) ->
        sources =
          case metadata["groundingChunks"] do
            chunks when is_list(chunks) ->
              Enum.map(chunks, fn chunk ->
                case chunk do
                  %{"web" => %{"uri" => uri, "title" => title}} ->
                    %{"uri" => uri, "title" => title}

                  %{"web" => %{"uri" => uri}} ->
                    %{"uri" => uri}

                  _ ->
                    nil
                end
              end)
              |> Enum.reject(&is_nil/1)

            _ ->
              []
          end

        %{
          "grounding_metadata" => metadata,
          "sources" => sources
        }

      _ ->
        nil
    end
  end

  defp extract_grounding_metadata(_), do: nil

  defp maybe_add_thinking_config(config, budget, level) do
    case {budget, level} do
      {b, l} when not is_nil(b) and not is_nil(l) ->
        raise ArgumentError,
              "google_thinking_budget and google_thinking_level cannot be combined in the same request"

      {nil, nil} ->
        config

      {0, nil} ->
        Map.put(config, :thinkingConfig, %{thinkingBudget: 0})

      {budget, nil} when is_integer(budget) and budget > 0 ->
        Map.put(config, :thinkingConfig, %{thinkingBudget: budget, includeThoughts: true})

      {nil, level} ->
        Map.put(config, :thinkingConfig, %{
          thinkingLevel: to_string(level),
          includeThoughts: true
        })
    end
  end

  defp convert_google_to_openai_format(%{"candidates" => candidates} = body) do
    choice =
      case List.first(candidates) do
        %{"content" => %{"parts" => parts}} = candidate ->
          {content_parts, has_thinking?} = convert_google_parts_to_content(parts)
          tool_calls = extract_tool_calls(parts)
          reasoning_details = extract_reasoning_details_from_parts(parts)

          message =
            if has_thinking? or tool_calls != [] do
              %{
                "role" => "assistant",
                "content" => content_parts
              }
            else
              text_content =
                content_parts
                |> Enum.filter(&(&1["type"] == "text"))
                |> Enum.map_join("", & &1["text"])

              %{
                "role" => "assistant",
                "content" => text_content
              }
            end

          message =
            case tool_calls do
              [] -> message
              _ -> Map.put(message, "tool_calls", tool_calls)
            end

          message =
            case reasoning_details do
              [] -> message
              details -> Map.put(message, "reasoning_details", details)
            end

          finish_reason =
            case {tool_calls, candidate["finishReason"]} do
              {[_ | _], "STOP"} -> "tool_calls"
              {_, reason} -> normalize_google_finish_reason(reason)
            end

          %{
            "message" => message,
            "finish_reason" => finish_reason
          }

        %{"content" => content, "finishReason" => finish_reason} when is_map(content) ->
          %{
            "message" => %{"role" => "assistant", "content" => ""},
            "finish_reason" => normalize_google_finish_reason(finish_reason)
          }

        _ ->
          %{
            "message" => %{"role" => "assistant", "content" => ""},
            "finish_reason" => "stop"
          }
      end

    %{
      "id" => body["id"] || "google-#{System.unique_integer([:positive])}",
      "choices" => [choice],
      "usage" => convert_google_usage(body["usageMetadata"])
    }
  end

  defp convert_google_to_openai_format(body) when is_map(body), do: body
  defp convert_google_to_openai_format(_body), do: %{}

  defp convert_google_json_mode_to_openai_format(%{"candidates" => candidates} = body) do
    choice =
      case List.first(candidates) do
        %{"content" => %{"parts" => parts}} = candidate ->
          json_text =
            parts
            |> Enum.filter(&Map.has_key?(&1, "text"))
            |> Enum.map_join("", & &1["text"])

          # Return as text content (like OpenAI json_schema mode)
          # ReqLLM.Response.unwrap_object will parse the JSON
          %{
            "message" => %{
              "role" => "assistant",
              "content" => json_text
            },
            "finish_reason" => normalize_google_finish_reason(candidate["finishReason"])
          }

        _ ->
          %{
            "message" => %{"role" => "assistant", "content" => ""},
            "finish_reason" => "stop"
          }
      end

    %{
      "id" => body["id"] || "google-#{System.unique_integer([:positive])}",
      "choices" => [choice],
      "usage" => convert_google_usage(body["usageMetadata"])
    }
  end

  defp convert_google_json_mode_to_openai_format(body) when is_map(body), do: body
  defp convert_google_json_mode_to_openai_format(_body), do: %{}

  defp convert_google_parts_to_content(parts) do
    content_parts =
      parts
      |> Enum.filter(&Map.has_key?(&1, "text"))
      |> Enum.map(fn part ->
        if Map.get(part, "thought", false) do
          %{"type" => "thinking", "thinking" => part["text"]}
        else
          %{"type" => "text", "text" => part["text"]}
        end
      end)

    has_thinking? = Enum.any?(content_parts, &(&1["type"] == "thinking"))
    {content_parts, has_thinking?}
  end

  defp extract_tool_calls(parts) do
    for %{"functionCall" => %{} = call} = part <- parts do
      call_id = Map.get(call, "id", "tool_call_#{System.unique_integer([:positive])}")

      encoded_args =
        call
        |> Map.get("args", %{})
        |> Jason.encode!()

      tc = %{
        "id" => call_id,
        "type" => "function",
        "function" => %{
          "name" => call["name"],
          "arguments" => encoded_args
        }
      }

      # Preserve thoughtSignature from Gemini response so consumers can
      # cache and round-trip it for multi-turn tool calling with thinking models
      case Map.get(part, "thoughtSignature") do
        nil -> tc
        sig -> Map.put(tc, "thought_signature", sig)
      end
    end
  end

  defp extract_reasoning_details_from_parts(parts) do
    parts
    |> Enum.filter(&(Map.get(&1, "thought", false) == true))
    |> Enum.with_index()
    |> Enum.map(fn {part, index} ->
      %ReqLLM.Message.ReasoningDetails{
        text: part["text"],
        signature: part["thoughtSignature"],
        encrypted?: part["thoughtSignature"] != nil,
        provider: :google,
        format: "google-gemini-v1",
        index: index,
        provider_data: %{"thought" => true}
      }
    end)
  end

  defp extract_reasoning_details_from_openai_format(%{"choices" => [first_choice | _]}) do
    case first_choice do
      %{"message" => %{"reasoning_details" => details}} when is_list(details) -> details
      _ -> nil
    end
  end

  defp extract_reasoning_details_from_openai_format(_), do: nil

  defp attach_reasoning_details(response, nil), do: response
  defp attach_reasoning_details(response, []), do: response

  defp attach_reasoning_details(%ReqLLM.Response{message: message} = response, details)
       when message != nil do
    updated_message = %{message | reasoning_details: details}

    updated_context =
      case Enum.split(response.context.messages, -1) do
        {init, [last]} when is_struct(last, ReqLLM.Message) and last.role == message.role ->
          updated_last = %{last | reasoning_details: details}
          %{response.context | messages: init ++ [updated_last]}

        _ ->
          response.context
      end

    %{response | message: updated_message, context: updated_context}
  end

  defp attach_reasoning_details(response, _details), do: response

  defp normalize_google_finish_reason("STOP"), do: "stop"
  defp normalize_google_finish_reason("MAX_TOKENS"), do: "length"
  defp normalize_google_finish_reason("SAFETY"), do: "content_filter"
  defp normalize_google_finish_reason("RECITATION"), do: "content_filter"
  defp normalize_google_finish_reason("OTHER"), do: "error"
  defp normalize_google_finish_reason(_), do: "error"

  defp convert_google_usage(usage) when is_map(usage) do
    normalized = normalize_google_usage(usage)

    base = %{
      "prompt_tokens" => normalized.input_tokens,
      "completion_tokens" => normalized.output_tokens,
      "total_tokens" => normalized.total_tokens
    }

    base =
      if normalized.reasoning_tokens > 0 do
        Map.put(base, "completion_tokens_details", %{
          "reasoning_tokens" => normalized.reasoning_tokens
        })
      else
        base
      end

    if normalized.cached_tokens > 0 do
      Map.put(base, "prompt_tokens_details", %{"cached_tokens" => normalized.cached_tokens})
    else
      base
    end
  end

  defp convert_google_usage(_),
    do: %{"prompt_tokens" => 0, "completion_tokens" => 0, "total_tokens" => 0}

  defp build_request_headers(_model, _opts), do: [{"Content-Type", "application/json"}]

  defp build_request_url(model_name, opts) do
    base_url = Keyword.fetch!(opts, :base_url)

    if use_header_auth?(opts) do
      "#{base_url}/models/#{model_name}:streamGenerateContent?alt=sse"
    else
      api_key = ReqLLM.Keys.get!(opts[:model_struct] || opts[:model], opts)
      "#{base_url}/models/#{model_name}:streamGenerateContent?key=#{api_key}&alt=sse"
    end
  end

  defp use_header_auth?(opts) do
    provider_options = Keyword.get(opts, :provider_options, [])
    Keyword.get(provider_options, :google_auth_header, false)
  end

  defp maybe_add_auth_header(headers, opts) do
    if use_header_auth?(opts) do
      api_key = ReqLLM.Keys.get!(opts[:model_struct] || opts[:model], opts)
      [{"x-goog-api-key", api_key} | headers]
    else
      headers
    end
  end

  defp build_request_body(model, context, opts) do
    operation = Keyword.get(opts, :operation, :chat)
    compiled_schema = Keyword.get(opts, :compiled_schema)

    base_options =
      [
        model: model.id,
        context: context,
        stream: true,
        operation: operation
      ]
      |> then(fn opts ->
        if compiled_schema, do: Keyword.put(opts, :compiled_schema, compiled_schema), else: opts
      end)

    all_options = Keyword.merge(base_options, Keyword.delete(opts, :finch_name))

    temp_request =
      Req.new(method: :post, url: URI.parse("https://example.com/temp"))
      |> Map.put(:options, Map.new(all_options))

    request_with_json = encode_body(temp_request)
    json_body = request_with_json.options[:json] || %{}
    Jason.encode_to_iodata!(json_body)
  end

  @impl ReqLLM.Provider
  def attach_stream(model, context, opts, _finch_name) do
    require Logger

    Logger.debug("Google attach_stream - model: #{inspect(model)}")

    req_only_keys = [
      :params,
      :model,
      :base_url,
      :finch_name,
      :fixture,
      :retry,
      :max_retries,
      :retry_log_level
    ]

    {req_opts, user_opts} = Keyword.split(opts, req_only_keys)

    operation = Keyword.get(user_opts, :operation, :chat)

    processed_opts0 =
      ReqLLM.Provider.Options.process_stream!(
        __MODULE__,
        operation,
        model,
        context,
        user_opts
      )

    with :ok <- validate_version_feature_compat(processed_opts0) do
      require Logger

      Logger.debug(
        "Google attach_stream - processed_opts0[:base_url]: #{inspect(processed_opts0[:base_url])}, api_version: #{inspect(resolve_api_version(processed_opts0))}"
      )

      computed_base_url = effective_base_url(processed_opts0)
      processed_opts = Keyword.put(processed_opts0, :base_url, computed_base_url)

      base_url = Keyword.get(req_opts, :base_url, processed_opts[:base_url])

      Logger.debug(
        "Google attach_stream - computed_base_url: #{inspect(computed_base_url)}, req_opts[:base_url]: #{inspect(req_opts[:base_url])}, final base_url: #{inspect(base_url)}"
      )

      opts_with_base = Keyword.merge(processed_opts, base_url: base_url, model_struct: model)

      base_headers =
        build_request_headers(model, opts_with_base) ++ [{"Accept", "text/event-stream"}]

      headers_with_auth = maybe_add_auth_header(base_headers, opts_with_base)
      custom_headers = ReqLLM.Provider.Utils.extract_custom_headers(opts[:req_http_options])
      headers = headers_with_auth ++ custom_headers
      url = build_request_url(model.id, opts_with_base)
      body = build_request_body(model, context, processed_opts)

      Logger.debug("Google attach_stream URL: #{inspect(sanitize_url(url))}")

      finch_request = Finch.build(:post, url, headers, body)
      {:ok, finch_request}
    end
  rescue
    error ->
      {:error,
       ReqLLM.Error.API.Request.exception(
         reason: "Failed to build Google stream request: #{inspect(error)}"
       )}
  end

  @impl ReqLLM.Provider
  def parse_stream_protocol(chunk, {:json_array, %{buffer: _buffer} = state}) do
    parse_json_array_protocol(chunk, state)
  end

  def parse_stream_protocol(chunk, {:json_array, buffer}) when is_binary(buffer) do
    parse_json_array_protocol(chunk, %{buffer: buffer, started?: true})
  end

  def parse_stream_protocol(chunk, state) do
    if json_array_protocol_start?(chunk, state) do
      state
      |> json_array_buffer()
      |> Kernel.<>(chunk)
      |> parse_json_array_protocol()
    else
      ReqLLM.Provider.parse_stream_protocol(chunk, state)
    end
  end

  @impl ReqLLM.Provider
  def decode_stream_event(event, model) do
    case event do
      %{data: data} when is_map(data) -> decode_google_event(data, model)
      data when is_map(data) -> decode_google_event(data, model)
      _ -> []
    end
  end

  # Split messages into system instruction and contents for Google Gemini
  defp split_messages_for_gemini(messages, model) do
    {system_msgs, chat_msgs} =
      Enum.split_with(messages, fn message ->
        case message do
          %{role: :system} -> true
          %{"role" => "system"} -> true
          %{"role" => :system} -> true
          %{role: "system"} -> true
          _ -> false
        end
      end)

    system_instruction =
      case system_msgs do
        [] ->
          nil

        system_messages ->
          combined_text =
            system_messages
            |> Enum.map_join("\n\n", &extract_text_content/1)

          %{parts: [%{text: combined_text}]}
      end

    contents = convert_messages_to_gemini(chat_msgs, model)

    {system_instruction, contents}
  end

  defp convert_messages_to_gemini(messages, model) do
    messages
    |> Enum.map(&convert_single_message_to_gemini(&1, model))
    |> merge_consecutive_roles()
  end

  defp convert_single_message_to_gemini(message, model) do
    raw_role =
      case message do
        %{role: role} -> role
        %{"role" => role} -> role
        _ -> "user"
      end

    role =
      case raw_role do
        :user -> "user"
        "user" -> "user"
        :assistant -> "model"
        "assistant" -> "model"
        :tool -> "user"
        "tool" -> "user"
        :system -> "user"
        "system" -> "user"
        other when is_binary(other) -> other
        other -> to_string(other)
      end

    raw_content =
      case message do
        %{content: content} -> content
        %{"content" => content} -> content
        _ -> ""
      end

    thought_parts = encode_reasoning_details_for_gemini(message)

    tool_result? = tool_result_message?(message)

    nest_multimodal? =
      tool_result? and multimodal_tool_result?(raw_content) and gemini_3_or_later?(model)

    content_parts =
      cond do
        nest_multimodal? ->
          []

        is_binary(raw_content) ->
          [%{text: raw_content}]

        is_list(raw_content) ->
          Enum.map(raw_content, &convert_content_part/1)

        true ->
          []
      end

    tool_call_parts =
      case message do
        %{"tool_calls" => tool_calls} when is_list(tool_calls) ->
          Enum.map(tool_calls, &convert_tool_call_to_function_call/1)

        %{tool_calls: tool_calls} when is_list(tool_calls) ->
          Enum.map(tool_calls, &convert_tool_call_to_function_call/1)

        _ ->
          []
      end

    tool_result_parts =
      if tool_result? do
        [build_tool_result_part(message, raw_content, nest_multimodal?)]
      else
        []
      end

    parts = thought_parts ++ content_parts ++ tool_call_parts ++ tool_result_parts

    %{role: role, parts: parts}
  end

  defp tool_result_message?(%{tool_call_id: _, role: "tool"}), do: true
  defp tool_result_message?(%{tool_call_id: _, role: :tool}), do: true
  defp tool_result_message?(%{"tool_call_id" => _, "role" => "tool"}), do: true
  defp tool_result_message?(_), do: false

  defp multimodal_tool_result?(content) when is_list(content) do
    Enum.any?(content, &multimodal_part?/1)
  end

  defp multimodal_tool_result?(_), do: false

  defp multimodal_part?(%ReqLLM.Message.ContentPart{type: type})
       when type in [:file, :image, :image_url],
       do: true

  defp multimodal_part?(%{type: type}) when type in [:file, :image, :image_url], do: true
  defp multimodal_part?(%{type: type}) when type in ["file", "image", "image_url"], do: true
  defp multimodal_part?(%{"type" => type}) when type in ["file", "image", "image_url"], do: true
  defp multimodal_part?(_), do: false

  # Gemini requires that consecutive messages with the same role are merged
  # into a single entry. This is critical for parallel tool calls: N separate
  # tool result messages (all mapped to role "user") must become one entry
  # with N functionResponse parts.
  defp merge_consecutive_roles([]), do: []

  defp merge_consecutive_roles([first | rest]) do
    {merged, last} =
      Enum.reduce(rest, {[], first}, fn
        %{role: role, parts: parts}, {acc, %{role: role} = current} ->
          {acc, %{current | parts: current.parts ++ parts}}

        entry, {acc, current} ->
          {acc ++ [current], entry}
      end)

    merged ++ [last]
  end

  defp encode_reasoning_details_for_gemini(message) do
    reasoning_details =
      case message do
        %{reasoning_details: details} when is_list(details) and details != [] -> details
        %{"reasoning_details" => details} when is_list(details) and details != [] -> details
        _ -> nil
      end

    case reasoning_details do
      nil ->
        []

      details ->
        details
        |> Enum.sort_by(& &1.index)
        |> Enum.flat_map(&encode_single_google_reasoning_detail/1)
    end
  end

  defp encode_single_google_reasoning_detail(
         %ReqLLM.Message.ReasoningDetails{provider: :google} = detail
       ) do
    part = %{text: detail.text || "", thought: true}
    part = if detail.signature, do: Map.put(part, :thoughtSignature, detail.signature), else: part
    [part]
  end

  defp encode_single_google_reasoning_detail(%ReqLLM.Message.ReasoningDetails{provider: provider}) do
    Logger.debug(
      "Skipping non-Google reasoning detail from provider: #{inspect(provider)} in Google request"
    )

    []
  end

  defp encode_single_google_reasoning_detail(_), do: []

  # Gemini 3 models require thoughtSignature on functionCall parts in conversation
  # history. When proxying through OpenAI-compatible format (which has no concept of
  # thought_signatures), the signatures are lost during round-trip conversion.
  # Inject Google's recommended dummy value for history originating from non-Gemini
  # sources. Real signatures are preferred when available (see thought_signature option).
  # See: https://ai.google.dev/gemini-api/docs/thought-signatures
  @thought_sig_dummy Base.encode64("skip_thought_signature_validator")

  defp convert_tool_call_to_function_call(
         %ReqLLM.ToolCall{
           type: "function",
           function: %{name: name, arguments: args}
         } = tool_call
       ) do
    %{
      functionCall: %{name: name, args: Jason.decode!(args)},
      thoughtSignature: tool_call_thought_signature(tool_call)
    }
  end

  defp convert_tool_call_to_function_call(
         %{
           "type" => "function",
           "function" => %{"name" => name, "arguments" => args}
         } = tool_call
       ) do
    %{
      functionCall: %{name: name, args: Jason.decode!(args)},
      thoughtSignature: tool_call_thought_signature(tool_call)
    }
  end

  defp convert_tool_call_to_function_call(
         %{
           type: "function",
           function: %{name: name, arguments: args}
         } = tool_call
       ) do
    %{
      functionCall: %{name: name, args: Jason.decode!(args)},
      thoughtSignature: tool_call_thought_signature(tool_call)
    }
  end

  defp convert_tool_call_to_function_call(_), do: nil

  defp tool_call_thought_signature(tool_call) do
    metadata = ReqLLM.ToolCall.metadata(tool_call)

    Map.get(metadata, :thought_signature) ||
      Map.get(metadata, "thought_signature") ||
      Map.get(tool_call, :thought_signature) ||
      Map.get(tool_call, "thought_signature") ||
      @thought_sig_dummy
  end

  defp extract_content_text(content) when is_binary(content), do: content

  defp extract_content_text(parts) when is_list(parts) do
    parts
    |> Enum.map_join("", fn
      %{"type" => "text", "text" => text} -> text
      %{type: :text, text: text} -> text
      %{type: "text", text: text} -> text
      text when is_binary(text) -> text
      _ -> ""
    end)
  end

  defp extract_content_text(_), do: ""

  defp build_tool_result_part(message, raw_content, false) do
    %{
      functionResponse: %{
        name: tool_result_name(message),
        response: tool_result_response(message, raw_content)
      }
    }
  end

  defp build_tool_result_part(message, raw_content, true) do
    media_parts =
      raw_content
      |> Enum.filter(&multimodal_part?/1)
      |> Enum.map(&convert_content_part/1)

    %{
      functionResponse: %{
        name: tool_result_name(message),
        response: tool_result_response(message, raw_content),
        parts: media_parts
      }
    }
  end

  defp tool_result_name(%{name: name}) when is_binary(name) and name != "", do: name
  defp tool_result_name(%{"name" => name}) when is_binary(name) and name != "", do: name
  defp tool_result_name(_), do: "unknown"

  defp tool_result_response(message, raw_content) do
    output = ReqLLM.ToolResult.output_from_message(message)

    cond do
      is_map(output) or is_list(output) ->
        output

      is_binary(output) ->
        %{content: output}

      output != nil ->
        %{content: to_string(output)}

      true ->
        %{content: extract_content_text(raw_content)}
    end
  end

  # Extract text content from a message for system instruction
  defp extract_text_content(%{content: content}) when is_binary(content), do: content
  defp extract_text_content(%{"content" => content}) when is_binary(content), do: content

  defp extract_text_content(%{content: parts}) when is_list(parts) do
    extract_parts_text(parts)
  end

  defp extract_text_content(%{"content" => parts}) when is_list(parts) do
    extract_parts_text(parts)
  end

  defp extract_text_content(content) when is_binary(content), do: content
  defp extract_text_content(_), do: ""

  defp extract_parts_text(parts) do
    parts
    |> Enum.map_join("", fn
      %{type: :text, content: text} -> text
      %{"type" => "text", "text" => text} -> text
      %{text: text} -> text
      %{"text" => text} -> text
      text when is_binary(text) -> text
      part -> to_string(part)
    end)
  end

  defp normalize_context_video_urls(%ReqLLM.Context{messages: messages} = context) do
    %{
      context
      | messages: Enum.map(messages, &normalize_message_video_urls/1)
    }
  end

  defp normalize_message_video_urls(%ReqLLM.Message{content: content} = message)
       when is_list(content) do
    %{
      message
      | content: Enum.map(content, &normalize_content_part_video_url/1)
    }
  end

  defp normalize_message_video_urls(message), do: message

  defp normalize_content_part_video_url(%ReqLLM.Message.ContentPart{
         type: :video_url,
         url: url,
         media_type: media_type,
         metadata: metadata
       }) do
    part = ReqLLM.Message.ContentPart.image_url(url, metadata)

    if is_binary(media_type) and media_type != "" do
      %{part | media_type: media_type}
    else
      part
    end
  end

  defp normalize_content_part_video_url(part), do: part

  defp convert_content_part(%{type: type} = part)
       when type in ["image_url", "video_url", :image_url, :video_url] do
    case content_part_url(part) do
      url when is_binary(url) ->
        convert_url_content_part(part, url)

      _ ->
        %{text: inspect(part)}
    end
  end

  defp convert_content_part(%{"type" => type} = part)
       when type in ["image_url", "video_url"] do
    case content_part_url(part) do
      url when is_binary(url) ->
        convert_url_content_part(part, url)

      _ ->
        %{text: inspect(part)}
    end
  end

  # Most specific patterns first (file, image, etc.) - for ContentPart structs
  defp convert_content_part(%{type: :file, data: data, media_type: media_type})
       when is_binary(data) do
    encoded_data = Base.encode64(data)

    %{
      inline_data: %{
        mime_type: media_type,
        data: encoded_data
      }
    }
  end

  # Specific text patterns
  defp convert_content_part(%{type: :text, content: text}), do: %{text: text}
  defp convert_content_part(%{"type" => "text", "text" => text}), do: %{text: text}

  # Generic catch-all patterns (must come after specific patterns)
  defp convert_content_part(%{text: text}) when is_binary(text), do: %{text: text}
  defp convert_content_part(%{"text" => text}) when is_binary(text), do: %{text: text}
  defp convert_content_part(text) when is_binary(text), do: %{text: text}

  defp convert_content_part(part), do: %{text: to_string(part)}

  defp convert_url_content_part(part, url) do
    cond do
      String.starts_with?(url, "data:") ->
        case String.split(url, ",", parts: 2) do
          [header, base64_data] ->
            mime_type =
              case Regex.run(~r/data:([^;]+)/, header) do
                [_, type] -> type
                _ -> "image/jpeg"
              end

            %{
              inline_data: %{
                mime_type: mime_type,
                data: base64_data
              }
            }

          _ ->
            %{text: "[Malformed data URI]"}
        end

      String.starts_with?(url, "http://") or String.starts_with?(url, "https://") ->
        build_file_data(part, url)

      String.starts_with?(url, "gs://") ->
        build_file_data(part, url)

      true ->
        %{text: "[Unsupported URL scheme: #{String.slice(url, 0, 20)}...]"}
    end
  end

  # Builds a fileData map, omitting mimeType when it cannot be reliably inferred.
  # YouTube and other extensionless URLs need mimeType omitted so Gemini can infer it.
  defp build_file_data(part, url) do
    mime_type = get_mime_type_from_part(part, url)

    file_data = %{fileUri: url}

    file_data =
      case mime_type do
        nil -> file_data
        "application/octet-stream" -> file_data
        known -> Map.put(file_data, :mimeType, known)
      end

    %{fileData: file_data}
  end

  # Helper to extract mime type from part metadata or infer from URL extension
  defp get_mime_type_from_part(part, url) do
    # Try metadata first (if passed through from ContentPart)
    case part do
      %{media_type: type} when is_binary(type) and type != "" -> type
      %{"media_type" => type} when is_binary(type) and type != "" -> type
      %{image_url: %{media_type: type}} when is_binary(type) and type != "" -> type
      %{"image_url" => %{"media_type" => type}} when is_binary(type) and type != "" -> type
      %{video_url: %{media_type: type}} when is_binary(type) and type != "" -> type
      %{"video_url" => %{"media_type" => type}} when is_binary(type) and type != "" -> type
      _ -> infer_mime_type_from_url(url)
    end
  end

  defp content_part_url(part) do
    part[:url] || part["url"] ||
      nested_url(part[:image_url] || part["image_url"]) ||
      nested_url(part[:video_url] || part["video_url"])
  end

  defp nested_url(%{url: url}) when is_binary(url), do: url
  defp nested_url(%{"url" => url}) when is_binary(url), do: url
  defp nested_url(_), do: nil

  defp infer_mime_type_from_url(url) do
    # Strip query params and get extension
    path = url |> URI.parse() |> Map.get(:path, "") |> to_string()

    case Path.extname(path) |> String.downcase() do
      ".jpg" -> "image/jpeg"
      ".jpeg" -> "image/jpeg"
      ".png" -> "image/png"
      ".gif" -> "image/gif"
      ".webp" -> "image/webp"
      ".pdf" -> "application/pdf"
      ".mp3" -> "audio/mpeg"
      ".mp4" -> "video/mp4"
      ".m4a" -> "audio/mp4"
      ".wav" -> "audio/wav"
      _ -> nil
    end
  end

  defp imagen_model_id?(id) when is_binary(id), do: String.contains?(id, "imagen")
  defp imagen_model_id?(_), do: false

  # Decode Google streaming events.
  #
  # Google's :streamGenerateContent endpoint returns JSON array format (not SSE) for 2.5 models.
  # This function handles both formats:
  # - SSE format: %{data: {...}}
  # - JSON array element: raw map from parsed JSON array
  defp decode_google_event(data, model) when is_map(data) do
    # Extract grounding metadata if present (for Google Search grounding)
    grounding_data = extract_grounding_metadata(data)
    provider_meta = if grounding_data, do: %{"google" => grounding_data}

    case data do
      %{
        "candidates" => [%{"content" => %{"parts" => parts}, "finishReason" => finish_reason} | _],
        "usageMetadata" => usage
      }
      when finish_reason != nil ->
        chunks = extract_chunks_from_parts(parts)

        meta = %{
          usage: convert_google_usage_for_streaming(usage),
          finish_reason: normalize_google_finish_reason(finish_reason),
          model: model.id,
          terminal?: true
        }

        meta = if provider_meta, do: Map.put(meta, :provider_meta, provider_meta), else: meta
        chunks ++ [ReqLLM.StreamChunk.meta(meta)]

      %{
        "candidates" => [%{"content" => %{"parts" => parts}, "finishReason" => finish_reason} | _]
      }
      when finish_reason != nil ->
        chunks = extract_chunks_from_parts(parts)

        meta = %{
          finish_reason: normalize_google_finish_reason(finish_reason),
          terminal?: true
        }

        meta = if provider_meta, do: Map.put(meta, :provider_meta, provider_meta), else: meta
        chunks ++ [ReqLLM.StreamChunk.meta(meta)]

      %{"candidates" => [%{"content" => %{"parts" => parts}} | _], "usageMetadata" => usage} ->
        chunks = extract_chunks_from_parts(parts)

        meta = %{
          usage: convert_google_usage_for_streaming(usage),
          model: model.id
        }

        meta = if provider_meta, do: Map.put(meta, :provider_meta, provider_meta), else: meta
        chunks ++ [ReqLLM.StreamChunk.meta(meta)]

      %{"candidates" => [%{"content" => %{"parts" => parts}} | _]} ->
        chunks = extract_chunks_from_parts(parts)

        if provider_meta do
          chunks ++ [ReqLLM.StreamChunk.meta(%{provider_meta: provider_meta})]
        else
          chunks
        end

      %{
        "candidates" => [%{"finishReason" => finish_reason} | _],
        "usageMetadata" => usage
      }
      when finish_reason != nil ->
        meta = %{
          usage: convert_google_usage_for_streaming(usage),
          finish_reason: normalize_google_finish_reason(finish_reason),
          model: model.id,
          terminal?: true
        }

        meta = if provider_meta, do: Map.put(meta, :provider_meta, provider_meta), else: meta
        [ReqLLM.StreamChunk.meta(meta)]

      %{"candidates" => [%{"finishReason" => finish_reason} | _]}
      when finish_reason != nil ->
        meta = %{
          finish_reason: normalize_google_finish_reason(finish_reason),
          terminal?: true
        }

        meta = if provider_meta, do: Map.put(meta, :provider_meta, provider_meta), else: meta
        [ReqLLM.StreamChunk.meta(meta)]

      %{"usageMetadata" => usage} ->
        meta = %{
          usage: convert_google_usage_for_streaming(usage),
          model: model.id,
          terminal?: true
        }

        meta = if provider_meta, do: Map.put(meta, :provider_meta, provider_meta), else: meta
        [ReqLLM.StreamChunk.meta(meta)]

      _ ->
        []
    end
  end

  defp json_array_protocol_start?(chunk, state) when state in [nil, ""] do
    chunk |> String.trim_leading() |> String.starts_with?("[")
  end

  defp json_array_protocol_start?(chunk, buffer) when is_binary(buffer) do
    (buffer <> chunk) |> String.trim_leading() |> String.starts_with?("[")
  end

  defp json_array_protocol_start?(_chunk, _state), do: false

  defp json_array_buffer(buffer) when is_binary(buffer), do: buffer
  defp json_array_buffer(_state), do: ""

  defp parse_json_array_protocol(data) do
    parse_json_array_protocol("", json_array_state(data, false))
  end

  defp parse_json_array_protocol(chunk, %{buffer: buffer} = state) do
    state = normalize_json_array_state(state)

    case drain_json_array_protocol(%{state | buffer: buffer <> chunk}, []) do
      {:ok, [], nil} -> {:ok, [], nil}
      {:ok, [], next_state} -> {:incomplete, {:json_array, next_state}}
      {:ok, events, nil} -> {:ok, events, nil}
      {:ok, events, next_state} -> {:ok, events, {:json_array, next_state}}
      {:error, reason} -> {:error, reason}
    end
  end

  defp json_array_state(buffer, started?) do
    %{buffer: buffer, started?: started?, pending_events: []}
  end

  defp normalize_json_array_state(state) do
    Map.put_new(state, :pending_events, [])
  end

  defp drain_json_array_protocol(%{buffer: buffer, started?: false} = state, events) do
    case String.trim_leading(buffer) do
      "" ->
        {:ok, Enum.reverse(events), %{state | buffer: ""}}

      <<?[, rest::binary>> ->
        drain_json_array_protocol(%{state | buffer: rest, started?: true}, events)

      _other ->
        {:error, :invalid_json_array_stream}
    end
  end

  defp drain_json_array_protocol(%{buffer: buffer, started?: true} = state, events) do
    buffer = trim_json_array_element_prefix(buffer)

    case buffer do
      "" ->
        {:ok, Enum.reverse(events), %{state | buffer: ""}}

      <<?], rest::binary>> ->
        if String.trim_leading(rest) == "" do
          {:ok, complete_json_array_events(state, events), nil}
        else
          {:error, :invalid_json_array_stream}
        end

      _other ->
        parse_json_array_element(buffer, state, events)
    end
  end

  defp parse_json_array_element(buffer, state, events) do
    case take_complete_json_container(buffer) do
      {:ok, json, rest} ->
        decode_json_array_element(json, rest, state, events)

      :incomplete ->
        {:ok, Enum.reverse(events), %{state | buffer: buffer}}

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp decode_json_array_element(json, rest, state, events) do
    case Jason.decode(json) do
      {:ok, event} when is_map(event) ->
        queue_json_array_event(event, rest, state, events)

      {:ok, _other} ->
        {:error, :invalid_json_array_stream}

      {:error, reason} ->
        {:error, {:invalid_json_array_stream, reason}}
    end
  end

  defp queue_json_array_event(event, rest, state, events) do
    wrapped_event = %{data: event}

    if json_array_terminal_event?(event) or state.pending_events != [] do
      pending_events = [wrapped_event | state.pending_events]
      drain_json_array_protocol(%{state | buffer: rest, pending_events: pending_events}, events)
    else
      drain_json_array_protocol(%{state | buffer: rest}, [wrapped_event | events])
    end
  end

  defp complete_json_array_events(state, events) do
    Enum.reverse(events) ++ Enum.reverse(state.pending_events)
  end

  defp json_array_terminal_event?(%{
         "candidates" => [%{"finishReason" => finish_reason} | _]
       })
       when finish_reason != nil,
       do: true

  defp json_array_terminal_event?(%{
         "candidates" => [%{"content" => %{"parts" => parts}} | _]
       })
       when is_list(parts) do
    false
  end

  defp json_array_terminal_event?(%{"usageMetadata" => _usage}), do: true

  defp json_array_terminal_event?(_event), do: false

  defp trim_json_array_element_prefix(buffer) do
    buffer
    |> String.trim_leading()
    |> drop_json_array_comma()
    |> String.trim_leading()
  end

  defp drop_json_array_comma(<<?,, rest::binary>>), do: rest
  defp drop_json_array_comma(buffer), do: buffer

  defp take_complete_json_container(<<?{, _rest::binary>> = buffer) do
    scan_json_container(buffer, buffer, 0, 0, false, false)
  end

  defp take_complete_json_container(<<?[, _rest::binary>> = buffer) do
    scan_json_container(buffer, buffer, 0, 0, false, false)
  end

  defp take_complete_json_container(_buffer), do: {:error, :invalid_json_array_stream}

  defp scan_json_container(_original, <<>>, _offset, _depth, _in_string?, _escaped?),
    do: :incomplete

  defp scan_json_container(original, <<_byte, rest::binary>>, offset, depth, true, true) do
    scan_json_container(original, rest, offset + 1, depth, true, false)
  end

  defp scan_json_container(original, <<?\\, rest::binary>>, offset, depth, true, false) do
    scan_json_container(original, rest, offset + 1, depth, true, true)
  end

  defp scan_json_container(original, <<?", rest::binary>>, offset, depth, true, false) do
    scan_json_container(original, rest, offset + 1, depth, false, false)
  end

  defp scan_json_container(original, <<_byte, rest::binary>>, offset, depth, true, false) do
    scan_json_container(original, rest, offset + 1, depth, true, false)
  end

  defp scan_json_container(original, <<?", rest::binary>>, offset, depth, false, false) do
    scan_json_container(original, rest, offset + 1, depth, true, false)
  end

  defp scan_json_container(original, <<byte, rest::binary>>, offset, depth, false, false)
       when byte in [?{, ?[] do
    scan_json_container(original, rest, offset + 1, depth + 1, false, false)
  end

  defp scan_json_container(original, <<byte, rest::binary>>, offset, depth, false, false)
       when byte in [?}, ?]] do
    case depth - 1 do
      0 ->
        length = offset + 1
        <<json::binary-size(length), remaining::binary>> = original
        {:ok, json, remaining}

      next_depth when next_depth > 0 ->
        scan_json_container(original, rest, offset + 1, next_depth, false, false)

      _other ->
        {:error, :invalid_json_array_stream}
    end
  end

  defp scan_json_container(original, <<_byte, rest::binary>>, offset, depth, false, false) do
    scan_json_container(original, rest, offset + 1, depth, false, false)
  end

  defp extract_chunks_from_parts(parts) do
    parts
    |> Enum.flat_map(fn part ->
      cond do
        Map.has_key?(part, "text") ->
          text = Map.get(part, "text")

          if text == "" do
            []
          else
            if Map.get(part, "thought", false) do
              signature = Map.get(part, "thoughtSignature")
              meta = if signature, do: %{signature: signature}, else: %{}
              [ReqLLM.StreamChunk.thinking(text, meta)]
            else
              [ReqLLM.StreamChunk.text(text)]
            end
          end

        Map.has_key?(part, "functionCall") ->
          call = part["functionCall"]
          name = call["name"]
          args = call["args"] || %{}
          call_id = Map.get(call, "id", "call_#{System.unique_integer([:positive])}")
          meta = %{id: call_id}
          # Preserve thoughtSignature from Gemini response for consumers that
          # need to cache and round-trip it (e.g., OpenAI-compatible proxies)
          meta =
            case Map.get(part, "thoughtSignature") do
              nil -> meta
              sig -> Map.put(meta, :thought_signature, sig)
            end

          [ReqLLM.StreamChunk.tool_call(name, args, meta)]

        true ->
          []
      end
    end)
  end

  defp convert_google_usage_for_streaming(nil),
    do: %{
      input_tokens: 0,
      output_tokens: 0,
      total_tokens: 0,
      cached_tokens: 0,
      reasoning_tokens: 0
    }

  defp convert_google_usage_for_streaming(usage_metadata) do
    normalize_google_usage(usage_metadata)
  end

  @impl ReqLLM.Provider
  def credential_missing?(%ReqLLM.Error.Invalid.Parameter{parameter: param}) do
    String.contains?(param, ":api_key") and
      String.contains?(param, "GOOGLE_API_KEY")
  end

  def credential_missing?(_), do: false
end
