# encoding: utf-8

#--
# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#++

module Cassandra
  module Protocol
    class CqlByteBuffer < Ione::ByteBuffer
      # @private
      MINUS = '-'.freeze
      # @private
      ZERO = '0'.freeze
      # @private
      DECIMAL_POINT = '.'.freeze
      # @private
      FLOAT_STRING_FORMAT = 'F'.freeze
      # @private
      NO_CHAR = ''.freeze

      def inspect
        "#<#{self.class.name}:0x#{object_id.to_s(16)} #{to_str.inspect}>"
      end

      def read_unsigned_byte
        read_byte
      rescue RangeError => e
        raise Errors::DecodingError, e.message, e.backtrace
      end

      def read_varint(len = bytesize, signed = true)
        bytes = read(len)
        n = 0
        bytes.each_byte do |b|
          n = (n << 8) | b
        end
        n -= 2**(bytes.length * 8) if signed && bytes.getbyte(0) & 0x80 == 0x80
        n
      rescue RangeError => e
        raise Errors::DecodingError, e.message, e.backtrace
      end

      def read_decimal(len = bytesize)
        scale = read_signed_int
        number_string = read_varint(len - 4).to_s
        if scale <= 0
          # Special case where the actual scale is positive; scale in the protocol is actually negative of
          # reality.
          BigDecimal(number_string + '0' * -scale)
        else
          if number_string.length <= scale
            if number_string.start_with?(MINUS)
              number_string = number_string[1, number_string.length - 1]
              fraction_string = MINUS + ZERO << DECIMAL_POINT
            else
              fraction_string = ZERO + DECIMAL_POINT
            end
            (scale - number_string.length).times { fraction_string << ZERO }
            fraction_string << number_string
          else
            fraction_string = number_string[0, number_string.length - scale]
            fraction_string << DECIMAL_POINT
            fraction_string <<
              number_string[number_string.length - scale, number_string.length]
          end
          BigDecimal(fraction_string)
        end
      rescue Errors::DecodingError => e
        raise Errors::DecodingError, e.message, e.backtrace
      end

      def read_long
        top, bottom = read(8).unpack(Formats::TWO_INTS_FORMAT)
        return (top << 32) | bottom if top <= 0x7fffffff
        top ^= 0xffffffff
        bottom ^= 0xffffffff
        -((top << 32) | bottom) - 1
      rescue RangeError => e
        raise Errors::DecodingError, e.message, e.backtrace
      end

      def read_double
        read(8).unpack(Formats::DOUBLE_FORMAT).first
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a double: #{e.message}", e.backtrace
      end

      def read_double_le
        read(8).unpack(Formats::DOUBLE_FORMAT_LE).first
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a double: #{e.message}", e.backtrace
      end

      def read_float
        read(4).unpack(Formats::FLOAT_FORMAT).first
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a float: #{e.message}", e.backtrace
      end

      def read_signed_int
        n = read_int
        return n if n <= 0x7fffffff
        n - 0xffffffff - 1
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode an int: #{e.message}", e.backtrace
      end

      def read_unsigned_int_le
        read(4).unpack(Formats::INT_FORMAT_LE).first
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode an int: #{e.message}", e.backtrace
      end

      def read_unsigned_short
        read_short
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a short: #{e.message}", e.backtrace
      end

      def read_unsigned_short_le
        read(2).unpack(Formats::SHORT_FORMAT_LE).first
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a short: #{e.message}", e.backtrace
      end

      def read_string
        length = read_unsigned_short
        string = read(length)
        string.force_encoding(::Encoding::UTF_8)
        string
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a string: #{e.message}", e.backtrace
      end

      def read_long_string
        length = read_signed_int
        string = read(length)
        string.force_encoding(::Encoding::UTF_8)
        string
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a long string: #{e.message}",
              e.backtrace
      end

      def read_uuid(impl = Uuid)
        impl.new(read_varint(16, false))
      rescue Errors::DecodingError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a UUID: #{e.message}", e.backtrace
      end

      def read_string_list
        size = read_unsigned_short
        Array.new(size) { read_string }
      end

      def read_bytes
        size = read_signed_int
        return nil if size & 0x80000000 == 0x80000000
        read(size)
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a bytes: #{e.message}", e.backtrace
      end

      def read_short_bytes
        read(read_unsigned_short)
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a short bytes: #{e.message}",
              e.backtrace
      end

      def read_option
        id = read_unsigned_short
        value = nil
        value = yield id, self if block_given?
        [id, value]
      end

      def read_inet
        size = read_byte
        ip_addr = IPAddr.new_ntoh(read(size))
        port = read_int
        [ip_addr, port]
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode an INET: #{e.message}",
              e.backtrace
      end

      def read_inet_addr
        size = read_byte
        IPAddr.new_ntoh(read(size))
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode an INET addr: #{e.message}",
              e.backtrace
      end

      def read_consistency
        index = read_unsigned_short
        if index >= CONSISTENCIES.size || CONSISTENCIES[index].nil?
          raise Errors::DecodingError, "Unknown consistency index #{index}"
        end
        CONSISTENCIES[index]
      end

      def read_string_map
        map = {}
        map_size = read_unsigned_short
        map_size.times do
          key = read_string
          map[key] = read_string
        end
        map
      end

      def read_bytes_map
        map = {}
        map_size = read_unsigned_short
        map_size.times do
          key = read_string
          map[key] = read_bytes
        end
        map
      end

      def read_reason_map
        # reason_map is new in v5. Starts with an int indicating the number of key-value pairs, followed by
        # the key-value pairs. Keys are inet's, values are short int's.
        map = {}
        map_size = read_int
        map_size.times do
          key = read_inet_addr
          map[key] = read_short
        end
        map
      end

      def read_string_multimap
        map = {}
        map_size = read_unsigned_short
        map_size.times do
          key = read_string
          map[key] = read_string_list
        end
        map
      end

      def read_smallint
        n = read_short
        return n if n <= 0x7fff
        n - 0xffff - 1
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a smallint: #{e.message}", e.backtrace
      end

      def read_tinyint
        n = read_byte
        return n if n <= 0x7f
        n - 0xff - 1
      rescue RangeError => e
        raise Errors::DecodingError,
              "Not enough bytes available to decode a tinyint: #{e.message}", e.backtrace
      end

      def read_vint
        n = read_byte

        # Bits are indexed in Integer in little-endian order
        bytes_to_read = 7.downto(0).take_while {|i| n[i] == 1}.size
        return n unless bytes_to_read > 0

        rv = n & (0xff >> bytes_to_read)
        1.upto(bytes_to_read) do |idx|
          new_byte = read_byte
          rv <<= 8
          rv |= (new_byte & 0xff)
        end

        rv
      rescue RangeError => e
        raise Errors::DecodingError, e.message, e.backtrace
      end

      def read_signed_vint
        Util.decode_zigzag(read_vint)
      end

      def append_tinyint(n)
        append([n].pack(Formats::CHAR_FORMAT))
      end

      def append_smallint(n)
        append_short(n)
      end

      def append_int(n)
        append([n].pack(Formats::INT_FORMAT))
      end

      def append_short(n)
        append([n].pack(Formats::SHORT_FORMAT))
      end

      def append_string(str)
        str = str.to_s
        append_short(str.bytesize)
        append(str)
      end

      def append_long_string(str)
        append_int(str.bytesize)
        append(str)
      end

      def append_uuid(uuid)
        v = uuid.value
        append_int((v >> 96) & 0xffffffff)
        append_int((v >> 64) & 0xffffffff)
        append_int((v >> 32) & 0xffffffff)
        append_int((v >> 0) & 0xffffffff)
      end

      def append_string_list(strs)
        append_short(strs.size)
        strs.each do |str|
          append_string(str)
        end
        self
      end

      def append_bytes(bytes)
        if bytes
          append_int(bytes.bytesize)
          append(bytes)
        else
          append_int(-1)
        end
      end

      def append_short_bytes(bytes)
        if bytes
          append_short(bytes.bytesize)
          append(bytes)
        else
          append_short(-1)
        end
      end

      def append_consistency(consistency)
        index = CONSISTENCIES.index(consistency)
        raise Errors::EncodingError, %(Unknown consistency "#{consistency}") if index.nil? || CONSISTENCIES[index].nil?
        append_short(index)
      end

      def append_string_map(map)
        append_short(map.size)
        map.each do |key, value|
          append_string(key)
          append_string(value)
        end
        self
      end

      def append_bytes_map(map)
        append_short(map.size)
        map.each do |key, value|
          append_string(key)
          append_bytes(value)
        end
        self
      end

      def append_long(n)
        top = n >> 32
        bottom = n & 0xffffffff
        append_int(top)
        append_int(bottom)
      end

      def append_varint(n)
        num = n
        bytes = []
        loop do
          bytes << (num & 0xff)
          num >>= 8
          break if (num == 0 || num == -1) && (bytes.last[7] == num[7])
        end
        append(bytes.reverse.pack(Formats::BYTES_FORMAT))
      end

      def append_decimal(n)
        str = n.to_s(FLOAT_STRING_FORMAT)
        size = str.index(DECIMAL_POINT)
        number_string = str.gsub(DECIMAL_POINT, NO_CHAR)

        num = number_string.to_i
        raw = self.class.new.append_varint(num)
        append_int(number_string.length - size)
        append(raw)
      end

      def append_double(n)
        append([n].pack(Formats::DOUBLE_FORMAT))
      end

      def append_float(n)
        append([n].pack(Formats::FLOAT_FORMAT))
      end

      def append_vint(n)
        send_bytes = Util.to_min_byte_array(n)
        send_cnt = send_bytes.length

        raise Errors::EncodingError, "Too many bytes (#{bytes_to_send.length}) to send!" if send_cnt > 8

        send_cnt_byte = (0xff << (8 - send_cnt)) & 0xff
        append([send_cnt_byte].pack(Formats::BYTES_FORMAT))
        append(send_bytes.pack(Formats::BYTES_FORMAT))
      end

      def append_signed_vint32(n)
        append_vint(Util.encode_zigzag32(n))
      end

      def append_signed_vint64(n)
        append_vint(Util.encode_zigzag64(n))
      end

      def eql?(other)
        other.eql?(to_str)
      end
      alias == eql?
    end
  end
end
