///|
/// Resource limits applied while decoding or constructing Avro binary data.
///
/// Limits are deliberately part of the public API: callers reading untrusted
/// streams should choose bounds appropriate for their protocol.
pub struct CodecLimits {
  max_depth : Int
  max_collection_items : Int
  max_bytes : Int
} derive(Debug, Eq)

///|
pub fn CodecLimits::new(
  max_depth? : Int = 64,
  max_collection_items? : Int = 1_000_000,
  max_bytes? : Int = 64 * 1024 * 1024,
) -> CodecLimits {
  {
    max_depth: if max_depth < 1 {
      1
    } else {
      max_depth
    },
    max_collection_items: if max_collection_items < 0 {
      0
    } else {
      max_collection_items
    },
    max_bytes: if max_bytes < 0 {
      0
    } else {
      max_bytes
    },
  }
}

///|
pub fn CodecLimits::max_depth(self : CodecLimits) -> Int {
  self.max_depth
}

///|
pub fn CodecLimits::max_collection_items(self : CodecLimits) -> Int {
  self.max_collection_items
}

///|
pub fn CodecLimits::max_bytes(self : CodecLimits) -> Int {
  self.max_bytes
}

///|
/// A growable Avro-binary writer. It uses Avro's little-endian float layout
/// and zigzag LEB128 integer representation.
pub struct Encoder {
  buffer : Buffer
  limits : CodecLimits
} derive(Debug)

///|
pub fn Encoder::new(limits? : CodecLimits = CodecLimits::new()) -> Encoder {
  { buffer: Buffer(), limits }
}

///|
pub fn Encoder::length(self : Encoder) -> Int {
  self.buffer.length()
}

///|
pub fn Encoder::to_bytes(self : Encoder) -> Bytes {
  self.buffer.to_bytes()
}

///|
fn Encoder::ensure_capacity(
  self : Encoder,
  extra : Int,
) -> Unit raise CodecError {
  if extra < 0 || extra > self.limits.max_bytes - self.buffer.length() {
    raise LimitExceeded(
      offset=self.buffer.length(),
      limit="maximum encoded byte size",
    )
  }
}

///|
pub fn Encoder::write_byte(
  self : Encoder,
  byte : Byte,
) -> Unit raise CodecError {
  self.ensure_capacity(1)
  self.buffer.write_byte(byte)
}

///|
pub fn Encoder::write_raw(
  self : Encoder,
  bytes : BytesView,
) -> Unit raise CodecError {
  self.ensure_capacity(bytes.length())
  self.buffer.write_bytes(bytes)
}

///|
fn zigzag_encode(value : Int64) -> UInt64 {
  let bits = value.reinterpret_as_uint64()
  let sign : UInt64 = if value < 0 { UInt64::lnot(0UL) } else { 0UL }
  (bits << 1) ^ sign
}

///|
fn zigzag_decode(value : UInt64) -> Int64 {
  let sign : UInt64 = if (value & 1UL) == 1UL { UInt64::lnot(0UL) } else { 0UL }
  ((value >> 1) ^ sign).reinterpret_as_int64()
}

///|
pub fn Encoder::write_long(
  self : Encoder,
  value : Int64,
) -> Unit raise CodecError {
  let mut encoded = zigzag_encode(value)
  while encoded >= 0x80UL {
    self.write_byte(((encoded & 0x7fUL) | 0x80UL).to_byte())
    encoded = encoded >> 7
  }
  self.write_byte(encoded.to_byte())
}

///|
pub fn Encoder::write_int(self : Encoder, value : Int) -> Unit raise CodecError {
  self.write_long(Int64::from_int(value))
}

///|
pub fn Encoder::write_float(
  self : Encoder,
  value : Float,
) -> Unit raise CodecError {
  self.write_raw(value.to_le_bytes())
}

///|
pub fn Encoder::write_double(
  self : Encoder,
  value : Double,
) -> Unit raise CodecError {
  self.write_raw(value.reinterpret_as_uint64().to_le_bytes())
}

///|
pub fn Encoder::write_bytes(
  self : Encoder,
  value : BytesView,
) -> Unit raise CodecError {
  self.write_long(Int64::from_int(value.length()))
  self.write_raw(value)
}

///|
pub fn Encoder::write_string(
  self : Encoder,
  value : StringView,
) -> Unit raise CodecError {
  self.write_bytes(@utf8.encode(value))
}

///|
/// A bounded cursor over an Avro binary byte sequence.
pub struct Decoder {
  input : Bytes
  mut offset : Int
  limits : CodecLimits
} derive(Debug)

///|
pub fn Decoder::new(
  input : Bytes,
  limits? : CodecLimits = CodecLimits::new(),
) -> Decoder raise CodecError {
  if input.length() > limits.max_bytes {
    raise LimitExceeded(offset=0, limit="maximum input byte size")
  }
  { input, offset: 0, limits }
}

///|
pub fn Decoder::offset(self : Decoder) -> Int {
  self.offset
}

///|
pub fn Decoder::remaining(self : Decoder) -> Int {
  self.input.length() - self.offset
}

///|
pub fn Decoder::limits(self : Decoder) -> CodecLimits {
  self.limits
}

///|
pub fn Decoder::ensure_finished(self : Decoder) -> Unit raise CodecError {
  if self.remaining() != 0 {
    raise TrailingData(offset=self.offset)
  }
}

///|
pub fn Decoder::read_byte(self : Decoder) -> Byte raise CodecError {
  if self.offset >= self.input.length() {
    raise UnexpectedEof(offset=self.offset)
  }
  let value = self.input[self.offset]
  self.offset += 1
  value
}

///|
pub fn Decoder::read_raw(
  self : Decoder,
  length : Int,
) -> Bytes raise CodecError {
  let start = self.offset
  if length < 0 {
    raise InvalidLength(offset=start, length=Int64::from_int(length))
  }
  if length > self.remaining() {
    raise UnexpectedEof(offset=start)
  }
  self.offset += length
  self.input[start:self.offset].to_owned()
}

///|
fn Decoder::read_unsigned_varint(self : Decoder) -> UInt64 raise CodecError {
  let start = self.offset
  let mut result : UInt64 = 0UL
  for index in 0..<10 {
    let byte = self.read_byte().to_int()
    if index == 9 && byte > 1 {
      raise InvalidEncoding(offset=start, message="varint exceeds 64 bits")
    }
    result = result |
      (UInt64::extend_uint((byte & 0x7f).reinterpret_as_uint()) << (index * 7))
    if (byte & 0x80) == 0 {
      return result
    }
  }
  raise InvalidEncoding(offset=start, message="varint exceeds 10 bytes")
}

///|
pub fn Decoder::read_long(self : Decoder) -> Int64 raise CodecError {
  zigzag_decode(self.read_unsigned_varint())
}

///|
pub fn Decoder::read_int(self : Decoder) -> Int raise CodecError {
  let start = self.offset
  let value = self.read_long()
  let narrowed = value.to_int()
  if Int64::from_int(narrowed) != value {
    raise InvalidEncoding(
      offset=start,
      message="int value is outside the 32-bit range",
    )
  }
  narrowed
}

///|
pub fn Decoder::read_float(self : Decoder) -> Float raise CodecError {
  let bytes = self.read_raw(4)
  let bits = bytes[0].to_int() |
    (bytes[1].to_int() << 8) |
    (bytes[2].to_int() << 16) |
    (bytes[3].to_int() << 24)
  Float::reinterpret_from_int(bits)
}

///|
pub fn Decoder::read_double(self : Decoder) -> Double raise CodecError {
  let bytes = self.read_raw(8)
  let mut bits : UInt64 = 0UL
  for index in 0..<8 {
    bits = bits |
      (
        UInt64::extend_uint(bytes[index].to_int().reinterpret_as_uint()) <<
        (index * 8)
      )
  }
  bits.reinterpret_as_double()
}

///|
fn Decoder::read_length(self : Decoder) -> Int raise CodecError {
  let start = self.offset
  let length = self.read_long()
  if length < 0 || length > Int64::from_int(self.limits.max_bytes) {
    raise InvalidLength(offset=start, length~)
  }
  length.to_int()
}

///|
pub fn Decoder::read_bytes(self : Decoder) -> Bytes raise CodecError {
  self.read_raw(self.read_length())
}

///|
pub fn Decoder::read_string(self : Decoder) -> String raise CodecError {
  let start = self.offset
  @utf8.decode(self.read_bytes()) catch {
    _ =>
      raise InvalidEncoding(offset=start, message="string is not valid UTF-8")
  }
}