///|
fn mismatch(schema : @schema.Schema, datum : Datum) -> Unit raise CodecError {
  raise TypeMismatch(expected=schema.kind_name(), actual=datum.type_name())
}

///|
fn checked_depth(decoder : Decoder, depth : Int) -> Unit raise CodecError {
  if depth >= decoder.limits().max_depth() {
    raise LimitExceeded(
      offset=decoder.offset(),
      limit="maximum schema nesting depth",
    )
  }
}

///|
fn write_boolean(encoder : Encoder, value : Bool) -> Unit raise CodecError {
  encoder.write_byte(if value { 1 } else { 0 })
}

///|
fn read_boolean(decoder : Decoder) -> Bool raise CodecError {
  let offset = decoder.offset()
  match decoder.read_byte() {
    0 => false
    1 => true
    _ => raise InvalidEncoding(offset~, message="boolean byte must be 0 or 1")
  }
}

///|
fn record_value(fields : Array[(String, Datum)], name : String) -> Datum? {
  for _, pair in fields {
    let (field_name, value) = pair
    if field_name == name {
      return Some(value)
    }
  }
  None
}

///|
fn is_record_field(fields : Array[@schema.Field], name : String) -> Bool {
  for field in fields {
    if field.name() == name {
      return true
    }
  }
  false
}

///|
fn checked_collection_count(
  decoder : Decoder,
  count : Int64,
) -> Int raise CodecError {
  let offset = decoder.offset()
  if count < 0 ||
    count > Int64::from_int(decoder.limits().max_collection_items()) {
    raise InvalidLength(offset~, length=count)
  }
  count.to_int()
}

///|
fn collection_block(
  decoder : Decoder,
  count : Int64,
) -> (Int, Int?) raise CodecError {
  if count >= 0 {
    (checked_collection_count(decoder, count), None)
  } else {
    if count == 0x8000000000000000UL.reinterpret_as_int64() {
      raise InvalidLength(offset=decoder.offset(), length=count)
    }
    let item_count = checked_collection_count(decoder, -count)
    let size_offset = decoder.offset()
    let declared_size = decoder.read_long()
    if declared_size < 0 ||
      declared_size > Int64::from_int(decoder.limits().max_bytes()) {
      raise InvalidLength(offset=size_offset, length=declared_size)
    }
    (item_count, Some(declared_size.to_int()))
  }
}

///|
fn check_block_size(
  decoder : Decoder,
  start : Int,
  declared_size : Int?,
) -> Unit raise CodecError {
  match declared_size {
    Some(size) =>
      if decoder.offset() - start != size {
        raise InvalidEncoding(
          offset=start,
          message="collection block size does not match its payload",
        )
      }
    None => ()
  }
}

///|
#warnings("-unnecessary_annotation")
fn write_schema(
  root : @schema.Schema,
  schema : @schema.Schema,
  datum : Datum,
  encoder : Encoder,
  depth : Int,
) -> Unit raise CodecError {
  if depth >= encoder.limits.max_depth() {
    raise LimitExceeded(
      offset=encoder.length(),
      limit="maximum schema nesting depth",
    )
  }
  match schema.kind() {
    @schema.SchemaKind::Primitive(@schema.Primitive::Null) =>
      match datum {
        Null => ()
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Primitive(@schema.Primitive::Boolean) =>
      match datum {
        Boolean(value) => write_boolean(encoder, value)
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Primitive(@schema.Primitive::Int) =>
      match datum {
        Int(value) => encoder.write_int(value)
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Primitive(@schema.Primitive::Long) =>
      match datum {
        Long(value) => encoder.write_long(value)
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Primitive(@schema.Primitive::Float) =>
      match datum {
        Float(value) => encoder.write_float(value)
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Primitive(@schema.Primitive::Double) =>
      match datum {
        Double(value) => encoder.write_double(value)
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Primitive(@schema.Primitive::Bytes) =>
      match datum {
        Bytes(value) => encoder.write_bytes(value)
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Primitive(@schema.Primitive::String) =>
      match datum {
        String(value) => encoder.write_string(value)
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Record(name=_, namespace_=_, aliases=_, fields~) =>
      match datum {
        Record(values) => {
          for field in fields {
            match record_value(values, field.name()) {
              Some(value) =>
                write_schema(root, field.schema(), value, encoder, depth + 1)
              None => raise MissingField(field=field.name())
            }
          }
          for _, pair in values {
            let (field_name, _) = pair
            if !is_record_field(fields, field_name) {
              raise InvalidEncoding(
                offset=encoder.length(),
                message="record contains a field absent from its schema: \{field_name}",
              )
            }
          }
        }
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Enum(
      name=_,
      namespace_=_,
      aliases=_,
      symbols~,
      default_symbol=_
    ) =>
      match datum {
        Enum(index, symbol) => {
          if index < 0 || index >= symbols.length() || symbols[index] != symbol {
            raise InvalidEncoding(
              offset=encoder.length(),
              message="enum index and symbol do not match its schema",
            )
          }
          encoder.write_int(index)
        }
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Array(items~) =>
      match datum {
        Array(values) => {
          if values.length() > encoder.limits.max_collection_items() {
            raise LimitExceeded(
              offset=encoder.length(),
              limit="maximum collection item count",
            )
          }
          if !values.is_empty() {
            encoder.write_long(Int64::from_int(values.length()))
            for value in values {
              write_schema(root, items, value, encoder, depth + 1)
            }
          }
          encoder.write_long(0L)
        }
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Map(values~) =>
      match datum {
        Map(entries) => {
          if entries.length() > encoder.limits.max_collection_items() {
            raise LimitExceeded(
              offset=encoder.length(),
              limit="maximum collection item count",
            )
          }
          if !entries.is_empty() {
            encoder.write_long(Int64::from_int(entries.length()))
            for key, value in entries {
              encoder.write_string(key)
              write_schema(root, values, value, encoder, depth + 1)
            }
          }
          encoder.write_long(0L)
        }
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Union(branches~) =>
      match datum {
        Union(index, value) => {
          if index < 0 || index >= branches.length() {
            raise InvalidEncoding(
              offset=encoder.length(),
              message="union branch index is outside schema",
            )
          }
          encoder.write_long(Int64::from_int(index))
          write_schema(root, branches[index], value, encoder, depth + 1)
        }
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Fixed(name=_, namespace_=_, aliases=_, size~) =>
      match datum {
        Fixed(value) => {
          if value.length() != size {
            raise InvalidEncoding(
              offset=encoder.length(),
              message="fixed datum has the wrong size",
            )
          }
          encoder.write_raw(value)
        }
        _ => mismatch(schema, datum)
      }
    @schema.SchemaKind::Named(name) =>
      match root.resolve_named(name) {
        Some(kind) =>
          write_schema(
            root,
            @schema.Schema::new(kind, name),
            datum,
            encoder,
            depth + 1,
          )
        None =>
          raise InvalidEncoding(
            offset=encoder.length(),
            message="unresolved named schema: \{name}",
          )
      }
  }
}

///|
#warnings("-unnecessary_annotation")
fn read_schema(
  root : @schema.Schema,
  schema : @schema.Schema,
  decoder : Decoder,
  depth : Int,
) -> Datum raise CodecError {
  checked_depth(decoder, depth)
  match schema.kind() {
    @schema.SchemaKind::Primitive(@schema.Primitive::Null) => Null
    @schema.SchemaKind::Primitive(@schema.Primitive::Boolean) =>
      Boolean(read_boolean(decoder))
    @schema.SchemaKind::Primitive(@schema.Primitive::Int) =>
      Int(decoder.read_int())
    @schema.SchemaKind::Primitive(@schema.Primitive::Long) =>
      Long(decoder.read_long())
    @schema.SchemaKind::Primitive(@schema.Primitive::Float) =>
      Float(decoder.read_float())
    @schema.SchemaKind::Primitive(@schema.Primitive::Double) =>
      Double(decoder.read_double())
    @schema.SchemaKind::Primitive(@schema.Primitive::Bytes) =>
      Bytes(decoder.read_bytes())
    @schema.SchemaKind::Primitive(@schema.Primitive::String) =>
      String(decoder.read_string())
    @schema.SchemaKind::Record(name=_, namespace_=_, aliases=_, fields~) => {
      let values : Array[(String, Datum)] = []
      for field in fields {
        values.push(
          (field.name(), read_schema(root, field.schema(), decoder, depth + 1)),
        )
      }
      Record(values)
    }
    @schema.SchemaKind::Enum(
      name=_,
      namespace_=_,
      aliases=_,
      symbols~,
      default_symbol=_
    ) => {
      let offset = decoder.offset()
      let index = decoder.read_int()
      if index < 0 || index >= symbols.length() {
        raise InvalidEncoding(offset~, message="enum index is outside schema")
      }
      Enum(index, symbols[index])
    }
    @schema.SchemaKind::Array(items~) => {
      let result : Array[Datum] = []
      let mut count = decoder.read_long()
      while count != 0L {
        let (item_count, declared_size) = collection_block(decoder, count)
        if item_count >
          decoder.limits().max_collection_items() - result.length() {
          raise LimitExceeded(
            offset=decoder.offset(),
            limit="maximum collection item count",
          )
        }
        let start = decoder.offset()
        for _ in 0.. {
      let result : Map[String, Datum] = Map([])
      let mut count = decoder.read_long()
      while count != 0L {
        let (entry_count, declared_size) = collection_block(decoder, count)
        if entry_count >
          decoder.limits().max_collection_items() - result.length() {
          raise LimitExceeded(
            offset=decoder.offset(),
            limit="maximum collection item count",
          )
        }
        let start = decoder.offset()
        for _ in 0.. {
      let offset = decoder.offset()
      let index = decoder.read_int()
      if index < 0 || index >= branches.length() {
        raise InvalidEncoding(
          offset~,
          message="union branch index is outside schema",
        )
      }
      Union(index, read_schema(root, branches[index], decoder, depth + 1))
    }
    @schema.SchemaKind::Fixed(name=_, namespace_=_, aliases=_, size~) =>
      Fixed(decoder.read_raw(size))
    @schema.SchemaKind::Named(name) =>
      match root.resolve_named(name) {
        Some(kind) =>
          read_schema(root, @schema.Schema::new(kind, name), decoder, depth + 1)
        None =>
          raise InvalidEncoding(
            offset=decoder.offset(),
            message="unresolved named schema: \{name}",
          )
      }
  }
}

///|
/// Encode one datum according to an Avro schema.
pub fn encode(schema : @schema.Schema, datum : Datum) -> Bytes raise CodecError {
  let encoder = Encoder::new()
  write_schema(schema, schema, datum, encoder, 0)
  encoder.to_bytes()
}

///|
/// Decode one datum according to an Avro schema and reject trailing bytes.
pub fn decode(
  schema : @schema.Schema,
  input : Bytes,
  limits? : CodecLimits = CodecLimits::new(),
) -> Datum raise CodecError {
  let decoder = Decoder::new(input, limits~)
  let value = read_schema(schema, schema, decoder, 0)
  decoder.ensure_finished()
  value
}

///|
/// Encode one datum into a caller-owned stream encoder.
pub fn encode_into(
  schema : @schema.Schema,
  datum : Datum,
  encoder : Encoder,
) -> Unit raise CodecError {
  write_schema(schema, schema, datum, encoder, 0)
}

///|
/// Decode one datum from a caller-owned stream decoder, leaving later bytes
/// available for the next datum.
pub fn decode_from(
  schema : @schema.Schema,
  decoder : Decoder,
) -> Datum raise CodecError {
  read_schema(schema, schema, decoder, 0)
}