before_chto

defmodule Ecto.Adapters.ClickHouse.Schema do
  @moduledoc false
  @conn Ecto.Adapters.ClickHouse.Connection
  @dialyzer :no_improper_lists

  alias Ch.RowBinary

  # dialyzer complains that we pass {:raw, data} in query! params
  # TODO PR into Ecto to accept term as params instead of [term]
  @dialyzer {:no_fail_call, insert_all: 8, insert: 4}
  @dialyzer {:no_return, insert: 4}

  def insert_all(
        adapter_meta,
        schema_meta,
        header,
        rows,
        on_conflict,
        returning,
        placeholders,
        opts
      ) do
    %{source: source, prefix: prefix, schema: schema} = schema_meta
    opts = [{:command, :insert} | opts]

    %{num_rows: num_rows} =
      case rows do
        {%Ecto.Query{} = _query, params} ->
          sql = @conn.insert(prefix, source, header, rows, on_conflict, returning, placeholders)
          Ecto.Adapters.SQL.query!(adapter_meta, sql, params, opts)

        rows when is_list(rows) ->
          types = prepare_types(schema, header, opts)
          # TODO use RowBinaryWithNamesAndTypes for type discrepancy warnings
          sql = [@conn.insert(prefix, source, header, []) | " FORMAT RowBinary"]
          data = unzip_encode_rows(rows, header, types)
          Ecto.Adapters.SQL.query!(adapter_meta, sql, {:raw, data}, opts)
      end

    {num_rows, nil}
  end

  def insert(adapter_meta, schema_meta, params, opts) do
    %{source: source, prefix: prefix, schema: schema} = schema_meta
    {header, row} = :lists.unzip(params)

    types = prepare_types(schema, header, opts)
    sql = [@conn.insert(prefix, source, header, []) | " FORMAT RowBinary"]
    opts = [{:command, :insert} | opts]
    data = RowBinary.encode_row(row, types)

    Ecto.Adapters.SQL.query!(adapter_meta, sql, {:raw, data}, opts)
    {:ok, []}
  end

  def delete(adapter_meta, schema_meta, params, opts) do
    %{source: source, prefix: prefix} = schema_meta
    filter_values = Keyword.values(params)
    sql = @conn.delete(prefix, source, params, [])
    Ecto.Adapters.SQL.query!(adapter_meta, sql, filter_values, opts)
    {:ok, []}
  end

  defp extract_types(schema, fields) do
    Enum.map(fields, fn field ->
      type = schema.__schema__(:type, field) || raise "missing type for " <> inspect(field)
      type |> Ecto.Type.type() |> remap_type()
    end)
  end

  defp prepare_types(schema, header, opts) do
    cond do
      schema ->
        extract_types(schema, header)

      types = opts[:types] ->
        Enum.map(header, fn field -> Access.fetch!(types, field) end)

      true ->
        raise ArgumentError, "missing :types"
    end
  end

  defp remap_type(dt) when dt in [:naive_datetime, :utc_datetime], do: :datetime

  defp remap_type(usec) when usec in [:naive_datetime_usec, :utc_datetime_usec] do
    {:datetime64, :microsecond}
  end

  # TODO :integer is used in schema_versions schema
  defp remap_type(:integer), do: :i64
  defp remap_type(b) when b in [:binary_id, :binary], do: :string

  # Ch.Types.FixedString, Ch.Types.Nullable, etc.
  defp remap_type({:parameterized, :ch, type}), do: type
  defp remap_type({:array, type}), do: {:array, remap_type(type)}

  defp remap_type(other), do: other

  defp unzip_encode_rows([row | rows], header, types) do
    [unzip_encode_row(header, types, row) | unzip_encode_rows(rows, header, types)]
  end

  defp unzip_encode_rows([], _header, _types), do: []

  defp unzip_encode_row([field | fields], [type | types], row) do
    case :lists.keyfind(field, 1, row) do
      {_, value} -> [RowBinary.encode(type, value) | unzip_encode_row(fields, types, row)]
      false -> [RowBinary.encode(type, nil) | unzip_encode_row(fields, types, row)]
    end
  end

  defp unzip_encode_row([], [], _row), do: []
end