defmodule Xandra.Protocol.V5 do
  @moduledoc false

  import Bitwise

  import Xandra.Protocol,
    only: [decode_from_proto_type: 2, decode_from_proto_type: 3, encode_to_type: 2, is_decimal: 1]

  alias Xandra.{
    Batch,
    Error,
    Frame,
    Page,
    Prepared,
    Simple,
    TypeParser
  }

  alias Xandra.Protocol, as: Proto

  alias Xandra.Cluster.{StatusChange, TopologyChange}

  @unix_epoch_days 0x80000000

  @spec encode_request(Frame.t(kind), term, keyword) :: Frame.t(kind) when kind: var
  def encode_request(frame, params, options \\ [])

  def encode_request(%Frame{kind: :options} = frame, nil, _options) do
    %{frame | body: []}
  end

  def encode_request(%Frame{kind: :startup} = frame, requested_options, _options)
      when is_map(requested_options) do
    %Frame{frame | body: encode_to_type(requested_options, "[string map]")}
  end

  def encode_request(%Frame{kind: :auth_response} = frame, _requested_options, options) do
    case Keyword.fetch(options, :authentication) do
      {:ok, authentication} ->
        with {authenticator, auth_options} <- authentication,
             body <- authenticator.response_body(auth_options) do
          %{frame | body: [<<IO.iodata_length(body)::32>>, body]}
        else
          _ ->
            raise "the :authentication option must be " <>
                    "an {auth_module, auth_options} tuple, " <> "got: #{inspect(authentication)}"
        end

      :error ->
        raise "Cassandra server requires authentication but the :authentication option was not provided"
    end
  end

  def encode_request(%Frame{kind: :register} = frame, events, _options) when is_list(events) do
    %Frame{frame | body: encode_to_type(events, "[string list]")}
  end

  def encode_request(%Frame{kind: :query} = frame, %Simple{} = query, options) do
    %Simple{statement: statement, values: values, custom_payload: custom_payload} = query

    body = [
      encode_custom_payload(custom_payload),
      <<byte_size(statement)::32>>,
      statement,
      encode_params([], values, options, query.default_consistency, _skip_metadata? = false)
    ]

    %Frame{frame | body: body}
  end

  def encode_request(%Frame{kind: :prepare} = frame, %Prepared{} = prepared, _options) do
    %Prepared{statement: statement, request_custom_payload: custom_payload} = prepared

    %Frame{
      frame
      | body: [
          encode_custom_payload(custom_payload),
          <<byte_size(statement)::32>>,
          statement,
          <<0::32>>
        ]
    }
  end

  def encode_request(%Frame{kind: :execute} = frame, %Prepared{} = prepared, options) do
    %Prepared{
      id: id,
      bound_columns: columns,
      values: values,
      request_custom_payload: custom_payload
    } = prepared

    skip_metadata? = prepared.result_columns != nil

    body = [
      encode_custom_payload(custom_payload),
      <<byte_size(id)::16>>,
      id,
      <<byte_size(prepared.result_metadata_id)::16>>,
      prepared.result_metadata_id,
      encode_params(columns, values, options, prepared.default_consistency, skip_metadata?)
    ]

    %Frame{frame | body: body}
  end

  def encode_request(%Frame{kind: :batch} = frame, %Batch{} = batch, options) do
    %Batch{queries: queries, type: type, custom_payload: custom_payload} = batch

    consistency = Keyword.get(options, :consistency, batch.default_consistency)
    serial_consistency = Keyword.get(options, :serial_consistency)
    timestamp = Keyword.get(options, :timestamp)

    flags =
      0x00
      |> Proto.set_flag(_serial_consistency = 0x10, serial_consistency)
      |> Proto.set_flag(_default_timestamp = 0x20, timestamp)

    encoded_queries = [<<length(queries)::16>>] ++ Enum.map(queries, &encode_query_in_batch/1)

    body = [
      encode_custom_payload(custom_payload),
      encode_batch_type(type),
      encoded_queries,
      encode_to_type(consistency, "[consistency]"),
      <<flags::32>>,
      Proto.encode_serial_consistency(serial_consistency),
      if(timestamp, do: <<timestamp::64>>, else: [])
    ]

    %Frame{frame | body: body}
  end

  ## Server payloads
  # We want to have the code for encoding some payloads even if they're payloads that are
  # only server to client. This is pretty easy to write, but all-in-all very useful for
  # debugging and testing.

  def encode_request(%Frame{kind: :event} = frame, %_{} = event, _options) do
    %Frame{frame | body: Proto.encode_event(event)}
  end

  defp encode_custom_payload(nil) do
    []
  end

  defp encode_custom_payload(custom_payload) when is_map(custom_payload) do
    key_value_pairs =
      for {key, value} <- custom_payload do
        [<<byte_size(key)::16>>, key, <<byte_size(value)::32>>, value]
      end

    [<<map_size(custom_payload)::16>> | key_value_pairs]
  end

  defp encode_batch_type(:logged), do: 0
  defp encode_batch_type(:unlogged), do: 1
  defp encode_batch_type(:counter), do: 2

  defp encode_params(columns, values, options, default_consistency, skip_metadata?) do
    consistency = Keyword.get(options, :consistency, default_consistency)
    page_size = Keyword.get(options, :page_size, 10_000)
    paging_state = Keyword.get(options, :paging_state)
    serial_consistency = Keyword.get(options, :serial_consistency)
    timestamp = Keyword.get(options, :timestamp)

    flags =
      0x00
      |> Proto.set_query_values_flag(values)
      |> Proto.set_flag(_page_size = 0x04, true)
      |> Proto.set_flag(_metadata_presence = 0x02, skip_metadata?)
      |> Proto.set_flag(_paging_state = 0x08, paging_state)
      |> Proto.set_flag(_serial_consistency = 0x10, serial_consistency)
      |> Proto.set_flag(_default_timestamp = 0x20, timestamp)

    encoded_values =
      if values == [] or values == %{} do
        []
      else
        encode_query_values(columns, values)
      end

    [
      encode_to_type(consistency, "[consistency]"),
      <<flags::32>>,
      encoded_values,
      <<page_size::32>>,
      Proto.encode_paging_state(paging_state),
      Proto.encode_serial_consistency(serial_consistency),
      if(timestamp, do: <<timestamp::64>>, else: [])
    ]
  end

  defp encode_query_in_batch(%Simple{statement: statement, values: values}) do
    [
      _kind = 0,
      <<byte_size(statement)::32>>,
      statement,
      encode_query_values([], values)
    ]
  end

  defp encode_query_in_batch(%Prepared{id: id, bound_columns: bound_columns, values: values}) do
    [
      _kind = 1,
      <<byte_size(id)::16>>,
      id,
      encode_query_values(bound_columns, values)
    ]
  end

  defp encode_query_values([], values) when is_list(values) do
    [<<length(values)::16>>] ++ Enum.map(values, &encode_query_value/1)
  end

  defp encode_query_values([], values) when is_map(values) do
    parts =
      for {name, value} <- values do
        name = to_string(name)
        [<<byte_size(name)::16>>, name, encode_query_value(value)]
      end

    [<<map_size(values)::16>>] ++ parts
  end

  defp encode_query_values(columns, values) when is_list(values) do
    encode_bound_values(columns, values, [<<length(columns)::16>>])
  end

  defp encode_query_values(columns, values) when map_size(values) == length(columns) do
    parts =
      for {_keyspace, _table, name, type} <- columns do
        value = Map.fetch!(values, name)
        [<<byte_size(name)::16>>, name, encode_query_value(type, value)]
      end

    [<<map_size(values)::16>>] ++ parts
  end

  defp encode_bound_values([], [], acc) do
    acc
  end

  defp encode_bound_values([column | columns], [value | values], acc) do
    {_keyspace, _table, _name, type} = column
    acc = [acc | encode_query_value(type, value)]
    encode_bound_values(columns, values, acc)
  end

  defp encode_query_value({type, value}) when is_binary(type) do
    encode_query_value(TypeParser.parse(type), value)
  end

  defp encode_query_value(_type, nil) do
    <<-1::32>>
  end

  defp encode_query_value(_type, :not_set) do
    <<-2::32>>
  end

  defp encode_query_value(type, value) do
    acc = encode_value(type, value)
    [<<IO.iodata_length(acc)::32>>, acc]
  end

  defp encode_value(:ascii, value) when is_binary(value) do
    value
  end

  defp encode_value(:bigint, value) when is_integer(value) do
    <<value::64>>
  end

  defp encode_value(:blob, value) when is_binary(value) do
    value
  end

  defp encode_value(:boolean, value) do
    case value do
      true -> [1]
      false -> [0]
    end
  end

  defp encode_value(:counter, value) when is_integer(value) do
    <<value::64>>
  end

  defp encode_value(:date, %Date{} = value) do
    value = Proto.date_to_unix_days(value)
    <<value + @unix_epoch_days::32>>
  end

  defp encode_value(:date, value) when value in -@unix_epoch_days..(@unix_epoch_days - 1) do
    <<value + @unix_epoch_days::32>>
  end

  defp encode_value(:decimal, {value, scale}) do
    [encode_value(:int, scale), encode_value(:varint, value)]
  end

  # Decimal stores the decimal as "sign * coef * 10^exp", but Cassandra stores it
  # as "coef * 10^(-1 * exp).
  defp encode_value(:decimal, decimal) do
    if is_decimal(decimal) do
      %Decimal{coef: coef, exp: exp, sign: sign} = decimal
      encode_value(:decimal, {_value = sign * coef, _scale = -exp})
    else
      raise ArgumentError,
            "can only encode %Decimal{} structs or {value, scale} tuples as decimals"
    end
  end

  defp encode_value(:double, value) when is_float(value) do
    <<value::64-float>>
  end

  defp encode_value(:float, value) when is_float(value) do
    <<value::32-float>>
  end

  defp encode_value(:inet, {n1, n2, n3, n4}) do
    <<n1, n2, n3, n4>>
  end

  defp encode_value(:inet, {n1, n2, n3, n4, n5, n6, n7, n8}) do
    <<n1::16, n2::16, n3::16, n4::16, n5::16, n6::16, n7::16, n8::16>>
  end

  defp encode_value(:int, value) when is_integer(value) do
    <<value::32>>
  end

  defp encode_value({:list, [items_type]}, collection) when is_list(collection) do
    [<<length(collection)::32>>] ++ Enum.map(collection, &encode_query_value(items_type, &1))
  end

  defp encode_value({:map, [key_type, value_type]}, collection) when is_map(collection) do
    parts =
      for {key, value} <- collection do
        [
          encode_query_value(key_type, key),
          encode_query_value(value_type, value)
        ]
      end

    [<<map_size(collection)::32>>] ++ parts
  end

  defp encode_value({:set, inner_type}, %MapSet{} = collection) do
    encode_value({:list, inner_type}, MapSet.to_list(collection))
  end

  defp encode_value(:smallint, value) when is_integer(value) do
    <<value::16>>
  end

  defp encode_value(:time, %Time{} = time) do
    value = Proto.time_to_nanoseconds(time)
    <<value::64>>
  end

  defp encode_value(:time, value) when value in 0..86_399_999_999_999 do
    <<value::64>>
  end

  defp encode_value(:timestamp, %DateTime{} = value) do
    <<DateTime.to_unix(value, :millisecond)::64>>
  end

  defp encode_value(:timestamp, value) when is_integer(value) do
    <<value::64>>
  end

  defp encode_value(:tinyint, value) when is_integer(value) do
    <<value>>
  end

  defp encode_value({:udt, fields}, value) when is_map(value) do
    for {field_name, [field_type]} <- fields do
      encode_query_value(field_type, Map.get(value, field_name))
    end
  end

  defp encode_value(type, value) when type in [:uuid, :timeuuid] and is_binary(value) do
    case byte_size(value) do
      16 ->
        value

      36 ->
        <<
          part1::8-bytes,
          ?-,
          part2::4-bytes,
          ?-,
          part3::4-bytes,
          ?-,
          part4::4-bytes,
          ?-,
          part5::12-bytes
        >> = value

        <<
          decode_base16(part1)::4-bytes,
          decode_base16(part2)::2-bytes,
          decode_base16(part3)::2-bytes,
          decode_base16(part4)::2-bytes,
          decode_base16(part5)::6-bytes
        >>
    end
  end

  defp encode_value(type, value) when type in [:varchar, :text] and is_binary(value) do
    value
  end

  defp encode_value(:varint, value) when is_integer(value) do
    size = varint_byte_size(value)
    <<value::size(size)-unit(8)>>
  end

  defp encode_value({:tuple, types}, value) when length(types) == tuple_size(value) do
    for {type, item} <- Enum.zip(types, Tuple.to_list(value)), do: encode_query_value(type, item)
  end

  defp varint_byte_size(value) when value > 127 do
    1 + varint_byte_size(value >>> 8)
  end

  defp varint_byte_size(value) when value < -128 do
    varint_byte_size(-value - 1)
  end

  defp varint_byte_size(_value), do: 1

  @compile {:inline, decode_base16: 1}
  defp decode_base16(value) do
    Base.decode16!(value, case: :mixed)
  end

  @compile {:inline, encode_base16: 1}
  defp encode_base16(value) do
    Base.encode16(value, case: :lower)
  end

  error_codes = %{
    0x0000 => :server_failure,
    0x000A => :protocol_violation,
    0x0100 => :invalid_credentials,
    0x1000 => :unavailable,
    0x1001 => :overloaded,
    0x1002 => :bootstrapping,
    0x1003 => :truncate_failure,
    0x1100 => :write_timeout,
    0x1200 => :read_timeout,
    0x1300 => :read_failure,
    0x1400 => :function_failure,
    0x1500 => :write_failure,
    0x1600 => :cdc_write_failure,
    0x1700 => :cas_write_unknown,
    0x2000 => :invalid_syntax,
    0x2100 => :unauthorized,
    0x2200 => :invalid,
    0x2300 => :invalid_config,
    0x2400 => :already_exists,
    0x2500 => :unprepared
  }

  for {code, reason} <- error_codes do
    defp decode_error_reason(<<unquote(code)::32-signed, buffer::bytes>>) do
      {unquote(reason), buffer}
    end
  end

  defp decode_error_message(_reason, buffer) do
    decode_from_proto_type(message <- buffer, "[string]")
    _ = buffer
    message
  end

  @spec decode_response(Frame.t(:error), term) :: Error.t()
  @spec decode_response(Frame.t(:ready), nil) :: :ok
  @spec decode_response(Frame.t(:supported), nil) :: %{optional(String.t()) => [String.t()]}
  @spec decode_response(Frame.t(:result), Simple.t() | Prepared.t() | Batch.t()) ::
          {Xandra.result() | Prepared.t(), warnings :: [String.t()]}
  def decode_response(frame, query \\ nil, options \\ [])

  def decode_response(%Frame{kind: :error, body: body, warning: warning?}, _query, _options) do
    {warnings, body} = Proto.decode_warnings(body, warning?)
    {reason, buffer} = decode_error_reason(body)
    Error.new(reason, decode_error_message(reason, buffer), warnings)
  end

  def decode_response(%Frame{kind: :ready, body: <<>>}, nil, _options) do
    :ok
  end

  def decode_response(%Frame{kind: :supported, body: body}, nil, _options) do
    decode_from_proto_type(value <- body, "[string multimap]")
    <<>> = body
    value
  end

  def decode_response(%Frame{kind: :event, body: body}, nil, _options) do
    decode_from_proto_type(event <- body, "[string]")
    decode_from_proto_type(effect <- body, "[string]")
    decode_from_proto_type(address_and_port <- body, "[inet]")
    {address, port} = address_and_port
    <<>> = body

    case event do
      "STATUS_CHANGE" -> %StatusChange{effect: effect, address: address, port: port}
      "TOPOLOGY_CHANGE" -> %TopologyChange{effect: effect, address: address, port: port}
    end
  end

  def decode_response(
        %Frame{
          kind: :result,
          body: body,
          tracing: tracing?,
          custom_payload: custom_payload?,
          atom_keys?: atom_keys?,
          warning: warning?
        },
        %kind{} = query,
        options
      )
      when kind in [Simple, Prepared, Batch] do
    {body, tracing_id} = Proto.decode_tracing_id(body, tracing?)
    {warnings, body} = Proto.decode_warnings(body, warning?)
    {custom_payload, body} = Proto.decode_custom_payload(body, custom_payload?)

    result =
      decode_result_response(
        body,
        query,
        tracing_id,
        custom_payload,
        Keyword.put(options, :atom_keys?, atom_keys?)
      )

    {result, warnings}
  end

  # Void
  defp decode_result_response(
         <<0x0001::32-signed>>,
         _query,
         tracing_id,
         custom_payload,
         _options
       ) do
    %Xandra.Void{tracing_id: tracing_id, custom_payload: custom_payload}
  end

  # Page
  defp decode_result_response(
         <<0x0002::32-signed, buffer::bits>>,
         query,
         tracing_id,
         custom_payload,
         options
       ) do
    %Page{} = page = Proto.new_page(query)
    {page, buffer} = decode_metadata(buffer, page, Keyword.fetch!(options, :atom_keys?))
    columns = rewrite_column_types(page.columns, options)

    %Page{
      page
      | content: decode_page_content(buffer, columns),
        tracing_id: tracing_id,
        custom_payload: custom_payload
    }
  end

  # SetKeyspace
  defp decode_result_response(
         <<0x0003::32-signed, buffer::bits>>,
         _query,
         tracing_id,
         custom_payload,
         _options
       ) do
    decode_from_proto_type(keyspace <- buffer, "[string]")
    <<>> = buffer

    %Xandra.SetKeyspace{
      keyspace: keyspace,
      tracing_id: tracing_id,
      custom_payload: custom_payload
    }
  end

  # Prepared
  defp decode_result_response(
         <<0x0004::32-signed, buffer::bits>>,
         %Prepared{} = prepared,
         tracing_id,
         custom_payload,
         options
       ) do
    atom_keys? = Keyword.fetch!(options, :atom_keys?)
    decode_from_proto_type(id <- buffer, "[string]")
    decode_from_proto_type(result_metadata_id <- buffer, "[string]")
    {%{columns: bound_columns}, buffer} = decode_metadata_prepared(buffer, %Page{}, atom_keys?)
    {%{columns: result_columns}, <<>>} = decode_metadata(buffer, %Page{}, atom_keys?)

    %Prepared{
      prepared
      | id: id,
        bound_columns: bound_columns,
        result_columns: result_columns,
        tracing_id: tracing_id,
        response_custom_payload: custom_payload,
        result_metadata_id: result_metadata_id
    }
  end

  # SchemaChange
  defp decode_result_response(
         <<0x0005::32-signed, buffer::bits>>,
         _query,
         tracing_id,
         custom_payload,
         _options
       ) do
    decode_from_proto_type(effect <- buffer, "[string]")
    decode_from_proto_type(target <- buffer, "[string]")
    options = decode_change_options(buffer, target)

    %Xandra.SchemaChange{
      effect: effect,
      target: target,
      options: options,
      tracing_id: tracing_id,
      custom_payload: custom_payload
    }
  end

  defp rewrite_column_types(columns, options) do
    Enum.map(columns, fn {_, _, _, type} = column ->
      put_elem(column, 3, rewrite_type(type, options))
    end)
  end

  defp rewrite_type({parent_type, types}, options) do
    {parent_type, Enum.map(types, &rewrite_type(&1, options))}
  end

  defp rewrite_type(:date, options) do
    {:date, [Keyword.get(options, :date_format, :date)]}
  end

  defp rewrite_type(:time, options) do
    {:time, [Keyword.get(options, :time_format, :time)]}
  end

  defp rewrite_type(:timestamp, options) do
    {:timestamp, [Keyword.get(options, :timestamp_format, :datetime)]}
  end

  defp rewrite_type(:decimal, options) do
    {:decimal, [Keyword.get(options, :decimal_format, :tuple)]}
  end

  defp rewrite_type(:uuid, options) do
    {:uuid, [Keyword.get(options, :uuid_format, :string)]}
  end

  defp rewrite_type(:timeuuid, options) do
    {:timeuuid, [Keyword.get(options, :timeuuid_format, :string)]}
  end

  defp rewrite_type(type, _options), do: type

  defp decode_change_options(<<buffer::bits>>, "KEYSPACE") do
    decode_from_proto_type(keyspace <- buffer, "[string]")
    <<>> = buffer
    %{keyspace: keyspace}
  end

  defp decode_change_options(<<buffer::bits>>, target) when target in ["TABLE", "TYPE"] do
    decode_from_proto_type(keyspace <- buffer, "[string]")
    decode_from_proto_type(subject <- buffer, "[string]")
    <<>> = buffer
    %{keyspace: keyspace, subject: subject}
  end

  defp decode_change_options(<<buffer::bits>>, target) when target in ["FUNCTION", "AGGREGATE"] do
    decode_from_proto_type(keyspace <- buffer, "[string]")
    decode_from_proto_type(subject <- buffer, "[string]")
    decode_from_proto_type(values <- buffer, "[string list]")
    <<>> = buffer
    %{keyspace: keyspace, subject: subject, arguments: values}
  end

  defp decode_metadata_prepared(
         <<flags::4-bytes, column_count::32-signed, pk_count::32-signed, buffer::bits>>,
         page,
         atom_keys?
       ) do
    <<_::31, global_table_spec::1>> = flags

    # partition key bind indices are ignored as we do not support token-aware routing
    {_indices, buffer} = decode_pk_index(buffer, pk_count, [])

    cond do
      global_table_spec == 1 ->
        decode_from_proto_type(keyspace <- buffer, "[string]")
        decode_from_proto_type(table <- buffer, "[string]")

        {columns, buffer} =
          decode_columns(buffer, column_count, {keyspace, table}, atom_keys?, [])

        {%{page | columns: columns}, buffer}

      true ->
        {columns, buffer} = decode_columns(buffer, column_count, nil, atom_keys?, [])
        {%{page | columns: columns}, buffer}
    end
  end

  # metadate format from the "Rows" result response
  defp decode_metadata(
         <<flags::4-bytes, column_count::32-signed, buffer::bits>>,
         page,
         atom_keys?
       ) do
    <<_::29, no_metadata::1, has_more_pages::1, global_table_spec::1>> = flags
    {page, buffer} = decode_paging_state(buffer, page, has_more_pages)

    cond do
      no_metadata == 1 ->
        {page, buffer}

      global_table_spec == 1 ->
        decode_from_proto_type(keyspace <- buffer, "[string]")
        decode_from_proto_type(table <- buffer, "[string]")

        {columns, buffer} =
          decode_columns(buffer, column_count, {keyspace, table}, atom_keys?, [])

        {%{page | columns: columns}, buffer}

      true ->
        {columns, buffer} = decode_columns(buffer, column_count, nil, atom_keys?, [])
        {%{page | columns: columns}, buffer}
    end
  end

  # pk = partition key
  def decode_pk_index(buffer, 0, acc) do
    {Enum.reverse(acc), buffer}
  end

  def decode_pk_index(<<index::16-unsigned, buffer::bits>>, pk_count, acc) do
    decode_pk_index(buffer, pk_count - 1, [index | acc])
  end

  defp decode_paging_state(<<buffer::bits>>, page, 0) do
    {page, buffer}
  end

  defp decode_paging_state(<<buffer::bits>>, page, 1) do
    <<size::32, paging_state::size(size)-bytes, buffer::bits>> = buffer
    {%{page | paging_state: paging_state}, buffer}
  end

  defp decode_page_content(<<row_count::32-signed, buffer::bits>>, columns) do
    decode_page_content(buffer, row_count, columns, columns, [[]])
  end

  defp decode_page_content(<<>>, 0, columns, columns, [[] | acc]) do
    Enum.reverse(acc)
  end

  defp decode_page_content(<<buffer::bits>>, row_count, columns, [], [values | acc]) do
    decode_page_content(buffer, row_count - 1, columns, columns, [[], Enum.reverse(values) | acc])
  end

  defp decode_page_content(<<buffer::bits>>, row_count, columns, [{_, _, _, type} | rest], [
         values | acc
       ]) do
    decode_from_proto_type(value <- buffer, "[value]") do
      value = decode_value(value, type)
      values = [value | values]
      decode_page_content(buffer, row_count, columns, rest, [values | acc])
    end
  end

  defp decode_value(nil, _type), do: nil

  defp decode_value(<<value>>, :boolean), do: value != 0
  defp decode_value(<<value::signed>>, :tinyint), do: value
  defp decode_value(<<value::16-signed>>, :smallint), do: value

  defp decode_value(<<value::64>>, {:time, [format]}) do
    case format do
      :time -> Proto.time_from_nanoseconds(value)
      :integer -> value
    end
  end

  defp decode_value(<<value::64-signed>>, :bigint), do: value
  defp decode_value(<<value::64-signed>>, :counter), do: value

  defp decode_value(<<value::64-signed>>, {:timestamp, [format]}) do
    case format do
      :datetime -> DateTime.from_unix!(value, :millisecond)
      :integer -> value
    end
  end

  defp decode_value(<<value::32>>, {:date, [format]}) do
    unix_days = value - @unix_epoch_days

    case format do
      :date -> Proto.date_from_unix_days(unix_days)
      :integer -> unix_days
    end
  end

  defp decode_value(<<value::32-signed>>, :int), do: value
  defp decode_value(<<value::64-float>>, :double), do: value
  defp decode_value(<<value::32-float>>, :float), do: value

  defp decode_value(<<data::4-bytes>>, :inet) do
    <<n1, n2, n3, n4>> = data
    {n1, n2, n3, n4}
  end

  defp decode_value(<<data::16-bytes>>, :inet) do
    <<n1::16, n2::16, n3::16, n4::16, n5::16, n6::16, n7::16, n8::16>> = data
    {n1, n2, n3, n4, n5, n6, n7, n8}
  end

  defp decode_value(<<value::16-bytes>>, {uuid_type, [format]})
       when uuid_type in [:uuid, :timeuuid] do
    case format do
      :binary ->
        value

      :string ->
        <<part1::32, part2::16, part3::16, part4::16, part5::48>> = value

        encode_base16(<<part1::32>>) <>
          "-" <>
          encode_base16(<<part2::16>>) <>
          "-" <>
          encode_base16(<<part3::16>>) <>
          "-" <>
          encode_base16(<<part4::16>>) <>
          "-" <>
          encode_base16(<<part5::48>>)
    end
  end

  defp decode_value(<<scale::32-signed, data::bits>>, {:decimal, [format]}) do
    value = decode_value(data, :varint)

    case format do
      :tuple ->
        {value, scale}

      :decimal ->
        sign = if(value >= 0, do: 1, else: -1)
        Decimal.new(sign, _coefficient = abs(value), _exponent = -scale)
    end
  end

  defp decode_value(<<count::32-signed, data::bits>>, {:list, [type]}) do
    decode_value_list(data, count, type, [])
  end

  defp decode_value(<<count::32-signed, data::bits>>, {:map, [key_type, value_type]}) do
    decode_value_map_key(data, count, key_type, value_type, [])
  end

  defp decode_value(<<count::32-signed, data::bits>>, {:set, [type]}) do
    data
    |> decode_value_list(count, type, [])
    |> MapSet.new()
  end

  defp decode_value(<<value::bits>>, :ascii), do: value
  defp decode_value(<<value::bits>>, :blob), do: value
  defp decode_value(<<value::bits>>, :varchar), do: value

  # For legacy compatibility reasons, most non-string types support
  # "empty" values (that is a value with zero length).
  # An empty value is distinct from NULL, which is encoded with a negative length.
  defp decode_value(<<>>, _type), do: nil

  defp decode_value(<<data::bits>>, {:udt, fields}) do
    decode_value_udt(data, fields, [])
  end

  defp decode_value(<<data::bits>>, {:tuple, types}) do
    decode_value_tuple(data, types, [])
  end

  defp decode_value(<<data::bits>>, :varint) do
    size = bit_size(data)
    <<value::size(size)-signed>> = data
    value
  end

  defp decode_value_udt(<<>>, fields, acc) do
    for {field_name, _} <- fields, into: Map.new(acc), do: {field_name, nil}
  end

  defp decode_value_udt(<<buffer::bits>>, [{field_name, [field_type]} | rest], acc) do
    decode_from_proto_type(value <- buffer, "[value]") do
      value = decode_value(value, field_type)
      decode_value_udt(buffer, rest, [{field_name, value} | acc])
    end
  end

  defp decode_value_list(<<>>, 0, _type, acc) do
    Enum.reverse(acc)
  end

  defp decode_value_list(<<buffer::bits>>, count, type, acc) do
    decode_from_proto_type(item <- buffer, "[value]") do
      item = decode_value(item, type)
      decode_value_list(buffer, count - 1, type, [item | acc])
    end
  end

  defp decode_value_map_key(<<>>, 0, _key_type, _value_type, acc) do
    Map.new(acc)
  end

  defp decode_value_map_key(<<buffer::bits>>, count, key_type, value_type, acc) do
    decode_from_proto_type(key <- buffer, "[value]") do
      key = decode_value(key, key_type)
      decode_value_map_value(buffer, count, key_type, value_type, [key | acc])
    end
  end

  defp decode_value_map_value(<<buffer::bits>>, count, key_type, value_type, [key | acc]) do
    decode_from_proto_type(value <- buffer, "[value]") do
      value = decode_value(value, value_type)
      decode_value_map_key(buffer, count - 1, key_type, value_type, [{key, value} | acc])
    end
  end

  defp decode_value_tuple(<<buffer::bits>>, [type | types], acc) do
    decode_from_proto_type(item <- buffer, "[value]") do
      item = decode_value(item, type)
      decode_value_tuple(buffer, types, [item | acc])
    end
  end

  defp decode_value_tuple(<<>>, [], acc) do
    acc |> Enum.reverse() |> List.to_tuple()
  end

  defp decode_columns(<<buffer::bits>>, 0, _table_spec, _atom_keys?, acc) do
    {Enum.reverse(acc), buffer}
  end

  defp decode_columns(<<buffer::bits>>, column_count, nil, atom_keys?, acc) do
    decode_from_proto_type(keyspace <- buffer, "[string]")
    decode_from_proto_type(table <- buffer, "[string]")
    decode_from_proto_type(name <- buffer, "[string]")
    name = if atom_keys?, do: String.to_atom(name), else: name
    {type, buffer} = decode_type(buffer)
    entry = {keyspace, table, name, type}
    decode_columns(buffer, column_count - 1, nil, atom_keys?, [entry | acc])
  end

  defp decode_columns(<<buffer::bits>>, column_count, table_spec, atom_keys?, acc) do
    {keyspace, table} = table_spec
    decode_from_proto_type(name <- buffer, "[string]")
    name = if atom_keys?, do: String.to_atom(name), else: name
    {type, buffer} = decode_type(buffer)
    entry = {keyspace, table, name, type}
    decode_columns(buffer, column_count - 1, table_spec, atom_keys?, [entry | acc])
  end

  defp decode_type(<<0x0001::16, buffer::bits>>) do
    {:ascii, buffer}
  end

  defp decode_type(<<0x0002::16, buffer::bits>>) do
    {:bigint, buffer}
  end

  defp decode_type(<<0x0003::16, buffer::bits>>) do
    {:blob, buffer}
  end

  defp decode_type(<<0x0004::16, buffer::bits>>) do
    {:boolean, buffer}
  end

  defp decode_type(<<0x0005::16, buffer::bits>>) do
    {:counter, buffer}
  end

  defp decode_type(<<0x0006::16, buffer::bits>>) do
    {:decimal, buffer}
  end

  defp decode_type(<<0x0007::16, buffer::bits>>) do
    {:double, buffer}
  end

  defp decode_type(<<0x0008::16, buffer::bits>>) do
    {:float, buffer}
  end

  defp decode_type(<<0x0009::16, buffer::bits>>) do
    {:int, buffer}
  end

  defp decode_type(<<0x000B::16, buffer::bits>>) do
    {:timestamp, buffer}
  end

  defp decode_type(<<0x000C::16, buffer::bits>>) do
    {:uuid, buffer}
  end

  defp decode_type(<<0x000D::16, buffer::bits>>) do
    {:varchar, buffer}
  end

  defp decode_type(<<0x000E::16, buffer::bits>>) do
    {:varint, buffer}
  end

  defp decode_type(<<0x000F::16, buffer::bits>>) do
    {:timeuuid, buffer}
  end

  defp decode_type(<<0x0010::16, buffer::bits>>) do
    {:inet, buffer}
  end

  defp decode_type(<<0x0011::16, buffer::bits>>) do
    {:date, buffer}
  end

  defp decode_type(<<0x0012::16, buffer::bits>>) do
    {:time, buffer}
  end

  defp decode_type(<<0x0013::16, buffer::bits>>) do
    {:smallint, buffer}
  end

  defp decode_type(<<0x0014::16, buffer::bits>>) do
    {:tinyint, buffer}
  end

  defp decode_type(<<0x0020::16, buffer::bits>>) do
    {type, buffer} = decode_type(buffer)
    {{:list, [type]}, buffer}
  end

  defp decode_type(<<0x0021::16, buffer::bits>>) do
    {key_type, buffer} = decode_type(buffer)
    {value_type, buffer} = decode_type(buffer)
    {{:map, [key_type, value_type]}, buffer}
  end

  defp decode_type(<<0x0022::16, buffer::bits>>) do
    {type, buffer} = decode_type(buffer)
    {{:set, [type]}, buffer}
  end

  defp decode_type(<<0x0030::16, buffer::bits>>) do
    decode_from_proto_type(_keyspace <- buffer, "[string]")
    decode_from_proto_type(_name <- buffer, "[string]")
    <<count::16, buffer::bits>> = buffer
    decode_type_udt(buffer, count, [])
  end

  defp decode_type(<<0x0031::16, count::16, buffer::bits>>) do
    decode_type_tuple(buffer, count, [])
  end

  defp decode_type_udt(<<buffer::bits>>, 0, acc) do
    {{:udt, Enum.reverse(acc)}, buffer}
  end

  defp decode_type_udt(<<buffer::bits>>, count, acc) do
    decode_from_proto_type(field_name <- buffer, "[string]")
    {field_type, buffer} = decode_type(buffer)
    decode_type_udt(buffer, count - 1, [{field_name, [field_type]} | acc])
  end

  defp decode_type_tuple(<<buffer::bits>>, 0, acc) do
    {{:tuple, Enum.reverse(acc)}, buffer}
  end

  defp decode_type_tuple(<<buffer::bits>>, count, acc) do
    {type, buffer} = decode_type(buffer)
    decode_type_tuple(buffer, count - 1, [type | acc])
  end
end