///|
fn check_depth(limits : JsonLimits, depth : Int) -> Unit raise JsonError {
  if depth >= limits.max_depth() {
    raise LimitExceeded(limit="maximum schema nesting depth")
  }
}

///|
fn check_items(limits : JsonLimits, actual : Int) -> Unit raise JsonError {
  if actual > limits.max_collection_items() {
    raise LimitExceeded(limit="maximum collection item count")
  }
}

///|
fn check_utf8(limits : JsonLimits, value : String) -> Unit raise JsonError {
  if @utf8.encode(value).length() > limits.max_string_bytes() {
    raise LimitExceeded(limit="maximum JSON string byte length")
  }
}

///|
fn json_kind(value : Json) -> String {
  match value {
    Null => "null"
    True | False => "boolean"
    Number(_, repr=_) => "number"
    String(_) => "string"
    Array(_) => "array"
    Object(_) => "object"
  }
}

///|
fn datum_kind(value : @codec.Datum) -> String {
  match value {
    Null => "null"
    Boolean(_) => "boolean"
    Int(_) => "int"
    Long(_) => "long"
    Float(_) => "float"
    Double(_) => "double"
    Bytes(_) => "bytes"
    String(_) => "string"
    Array(_) => "array"
    Map(_) => "map"
    Record(_) => "record"
    Enum(_, _) => "enum"
    Union(_, _) => "union"
    Fixed(_) => "fixed"
  }
}

///|
fn[T] mismatch(expected : String, value : Json) -> T raise JsonError {
  raise TypeMismatch(expected~, actual=json_kind(value))
}

///|
fn[T] datum_mismatch(
  expected : String,
  value : @codec.Datum,
) -> T raise JsonError {
  raise TypeMismatch(expected~, actual=datum_kind(value))
}

///|
fn integer_from_number(number : Double, repr : String?) -> Int raise JsonError {
  match repr {
    Some(text) =>
      @strconv.from_str(text) catch {
        _ => raise InvalidNumber(expected="32-bit integer", value=number)
      }
    None => {
      let integer = number.to_int()
      if number.is_nan() || number.is_inf() || integer.to_double() != number {
        raise InvalidNumber(expected="32-bit integer", value=number)
      }
      integer
    }
  }
}

///|
fn long_from_number(number : Double, repr : String?) -> Int64 raise JsonError {
  match repr {
    Some(text) =>
      @strconv.from_str(text) catch {
        _ => raise InvalidNumber(expected="64-bit integer", value=number)
      }
    None => {
      let integer = number.to_int64()
      if number.is_nan() || number.is_inf() || integer.to_double() != number {
        raise InvalidNumber(expected="64-bit integer", value=number)
      }
      integer
    }
  }
}

///|
fn byte_string(value : Bytes) -> String {
  String::from_array(value.to_array().map(byte => byte.to_char()))
}

///|
fn parse_byte_string(
  value : String,
  limits : JsonLimits,
) -> Bytes raise JsonError {
  check_items(limits, value.length())
  let bytes : Array[Byte] = []
  for character in value {
    let code = character.to_int()
    if code > 255 {
      raise InvalidByteString(
        message="Avro bytes JSON strings may only contain code points 0 through 255",
      )
    }
    bytes.push(code.to_byte())
  }
  Bytes::from_array(bytes)
}

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

///|
fn field_by_name(
  fields : Array[@schema.Field],
  name : String,
) -> @schema.Field? {
  for field in fields {
    if field.name() == name || field.aliases().contains(name) {
      return Some(field)
    }
  }
  None
}

///|
fn union_tag(
  root : @schema.Schema,
  branch : @schema.Schema,
) -> String raise JsonError {
  match branch.kind() {
    Primitive(Null) => "null"
    Primitive(Boolean) => "boolean"
    Primitive(Int) => "int"
    Primitive(Long) => "long"
    Primitive(Float) => "float"
    Primitive(Double) => "double"
    Primitive(Bytes) => "bytes"
    Primitive(String) => "string"
    Array(items=_) => "array"
    Map(values=_) => "map"
    Record(name~, namespace_=_, aliases=_, fields=_) => name
    Enum(name~, namespace_=_, aliases=_, symbols=_, default_symbol=_) => name
    Fixed(name~, namespace_=_, aliases=_, size=_) => name
    Union(_) =>
      raise InvalidUnion(
        message="a union cannot directly contain another union",
      )
    Named(name) =>
      match root.resolve_named(name) {
        Some(kind) => union_tag(root, @schema.Schema::new(kind, name))
        None => raise UnresolvedName(name)
      }
  }
}

///|
fn union_index(
  root : @schema.Schema,
  branches : Array[@schema.Schema],
  tag : String,
) -> Int? raise JsonError {
  for index, branch in branches {
    if union_tag(root, branch) == tag {
      return Some(index)
    }
  }
  None
}

///|
fn to_json_schema(
  root : @schema.Schema,
  schema : @schema.Schema,
  datum : @codec.Datum,
  limits : JsonLimits,
  depth : Int,
) -> Json raise JsonError {
  check_depth(limits, depth)
  match schema.kind() {
    Primitive(Null) =>
      match datum {
        Null => Json::null()
        _ => datum_mismatch("null", datum)
      }
    Primitive(Boolean) =>
      match datum {
        Boolean(value) => Json::boolean(value)
        _ => datum_mismatch("boolean", datum)
      }
    Primitive(Int) =>
      match datum {
        Int(value) => Json::number(value.to_double())
        _ => datum_mismatch("int", datum)
      }
    Primitive(Long) =>
      match datum {
        Long(value) => Json::number(value.to_double(), repr=value.to_string())
        _ => datum_mismatch("long", datum)
      }
    Primitive(Float) =>
      match datum {
        Float(value) => Json::number(value.to_double())
        _ => datum_mismatch("float", datum)
      }
    Primitive(Double) =>
      match datum {
        Double(value) => Json::number(value)
        _ => datum_mismatch("double", datum)
      }
    Primitive(Bytes) =>
      match datum {
        Bytes(value) => {
          check_items(limits, value.length())
          Json::string(byte_string(value))
        }
        _ => datum_mismatch("bytes", datum)
      }
    Primitive(String) =>
      match datum {
        String(value) => {
          check_utf8(limits, value)
          Json::string(value)
        }
        _ => datum_mismatch("string", datum)
      }
    Record(name=_, namespace_=_, aliases=_, fields~) =>
      match datum {
        Record(values) => {
          check_items(limits, values.length())
          let object : Map[String, Json] = Map([])
          for field in fields {
            match record_field(values, field.name()) {
              Some(value) =>
                object[field.name()] = to_json_schema(
                  root,
                  field.schema(),
                  value,
                  limits,
                  depth + 1,
                )
              None => raise MissingField(field=field.name())
            }
          }
          for _, pair in values {
            let (name, _) = pair
            if field_by_name(fields, name) is None {
              raise UnexpectedField(field=name)
            }
          }
          Json::object(object)
        }
        _ => datum_mismatch("record", datum)
      }
    Enum(name=_, namespace_=_, aliases=_, symbols~, default_symbol=_) =>
      match datum {
        Enum(index, symbol) => {
          if index < 0 || index >= symbols.length() || symbols[index] != symbol {
            raise InvalidEnum(symbol~)
          }
          Json::string(symbol)
        }
        _ => datum_mismatch("enum", datum)
      }
    Array(items~) =>
      match datum {
        Array(values) => {
          check_items(limits, values.length())
          Json::array(
            values.map(value => {
              to_json_schema(root, items, value, limits, depth + 1)
            }),
          )
        }
        _ => datum_mismatch("array", datum)
      }
    Map(values~) =>
      match datum {
        Map(entries) => {
          check_items(limits, entries.length())
          let object : Map[String, Json] = Map([])
          for key, value in entries {
            check_utf8(limits, key)
            object[key] = to_json_schema(root, values, value, limits, depth + 1)
          }
          Json::object(object)
        }
        _ => datum_mismatch("map", datum)
      }
    Union(branches~) =>
      match datum {
        Union(index, value) => {
          if index < 0 || index >= branches.length() {
            raise InvalidUnion(message="union branch index is outside schema")
          }
          let branch = branches[index]
          let encoded = to_json_schema(root, branch, value, limits, depth + 1)
          if branch.kind() is Primitive(Null) {
            encoded
          } else {
            let object : Map[String, Json] = Map([])
            object[union_tag(root, branch)] = encoded
            Json::object(object)
          }
        }
        _ => datum_mismatch("union", datum)
      }
    Fixed(name=_, namespace_=_, aliases=_, size~) =>
      match datum {
        Fixed(value) => {
          if value.length() != size {
            raise InvalidFixed(expected=size, actual=value.length())
          }
          Json::string(byte_string(value))
        }
        _ => datum_mismatch("fixed", datum)
      }
    Named(name) =>
      match root.resolve_named(name) {
        Some(kind) =>
          to_json_schema(
            root,
            @schema.Schema::new(kind, name),
            datum,
            limits,
            depth + 1,
          )
        None => raise UnresolvedName(name)
      }
  }
}

///|
fn from_json_schema(
  root : @schema.Schema,
  schema : @schema.Schema,
  value : Json,
  limits : JsonLimits,
  depth : Int,
) -> @codec.Datum raise JsonError {
  check_depth(limits, depth)
  match schema.kind() {
    Primitive(Null) =>
      if value is Null {
        Null
      } else {
        mismatch("null", value)
      }
    Primitive(Boolean) =>
      match value {
        True => Boolean(true)
        False => Boolean(false)
        _ => mismatch("boolean", value)
      }
    Primitive(Int) =>
      match value {
        Number(number, repr~) => Int(integer_from_number(number, repr))
        _ => mismatch("int", value)
      }
    Primitive(Long) =>
      match value {
        Number(number, repr~) => Long(long_from_number(number, repr))
        _ => mismatch("long", value)
      }
    Primitive(Float) =>
      match value {
        Number(number, repr=_) => Float(Float::from_double(number))
        _ => mismatch("float", value)
      }
    Primitive(Double) =>
      match value {
        Number(number, repr=_) => Double(number)
        _ => mismatch("double", value)
      }
    Primitive(Bytes) =>
      match value {
        String(text) => Bytes(parse_byte_string(text, limits))
        _ => mismatch("bytes", value)
      }
    Primitive(String) =>
      match value {
        String(text) => {
          check_utf8(limits, text)
          String(text)
        }
        _ => mismatch("string", value)
      }
    Record(name=_, namespace_=_, aliases=_, fields~) =>
      match value {
        Object(object) => {
          check_items(limits, object.length())
          let result : Array[(String, @codec.Datum)] = []
          for field in fields {
            let candidate = match object.get(field.name()) {
              Some(found) => Some(found)
              None => {
                let mut found : Json? = None
                for alternate in field.aliases() {
                  match object.get(alternate) {
                    Some(alias_value) => found = Some(alias_value)
                    None => ()
                  }
                }
                found
              }
            }
            match candidate {
              Some(field_value) =>
                result.push(
                  (
                    field.name(),
                    from_json_schema(
                      root,
                      field.schema(),
                      field_value,
                      limits,
                      depth + 1,
                    ),
                  ),
                )
              None =>
                match field.default_value() {
                  Some(default) =>
                    result.push(
                      (
                        field.name(),
                        from_json_schema(
                          root,
                          field.schema(),
                          default,
                          limits,
                          depth + 1,
                        ),
                      ),
                    )
                  None => raise MissingField(field=field.name())
                }
            }
          }
          for key, _ in object {
            if field_by_name(fields, key) is None {
              raise UnexpectedField(field=key)
            }
          }
          Record(result)
        }
        _ => mismatch("record", value)
      }
    Enum(name=_, namespace_=_, aliases=_, symbols~, default_symbol=_) =>
      match value {
        String(symbol) => {
          let mut index = -1
          for candidate_index, candidate in symbols {
            if candidate == symbol {
              index = candidate_index
            }
          }
          if index < 0 {
            raise InvalidEnum(symbol~)
          }
          Enum(index, symbol)
        }
        _ => mismatch("enum", value)
      }
    Array(items~) =>
      match value {
        Array(values) => {
          check_items(limits, values.length())
          Array(
            values.map(item => {
              from_json_schema(root, items, item, limits, depth + 1)
            }),
          )
        }
        _ => mismatch("array", value)
      }
    Map(values~) =>
      match value {
        Object(object) => {
          check_items(limits, object.length())
          let result : Map[String, @codec.Datum] = Map([])
          for key, item in object {
            check_utf8(limits, key)
            result[key] = from_json_schema(
              root,
              values,
              item,
              limits,
              depth + 1,
            )
          }
          Map(result)
        }
        _ => mismatch("map", value)
      }
    Union(branches~) =>
      if value is Null {
        match union_index(root, branches, "null") {
          Some(index) => Union(index, Null)
          None =>
            raise InvalidUnion(message="null does not occur in this union")
        }
      } else {
        match value {
          Object(object) => {
            if object.length() != 1 {
              raise InvalidUnion(
                message="a non-null union value must be an object with one branch key",
              )
            }
            for tag, branch_value in object {
              match union_index(root, branches, tag) {
                Some(index) =>
                  return Union(
                    index,
                    from_json_schema(
                      root,
                      branches[index],
                      branch_value,
                      limits,
                      depth + 1,
                    ),
                  )
                None =>
                  raise InvalidUnion(message="unknown union branch: \\{tag}")
              }
            }
            raise InvalidUnion(
              message="a non-null union value must contain one branch",
            )
          }
          _ =>
            raise InvalidUnion(
              message="a non-null union value must be an object",
            )
        }
      }
    Fixed(name=_, namespace_=_, aliases=_, size~) =>
      match value {
        String(text) => {
          let bytes = parse_byte_string(text, limits)
          if bytes.length() != size {
            raise InvalidFixed(expected=size, actual=bytes.length())
          }
          Fixed(bytes)
        }
        _ => mismatch("fixed", value)
      }
    Named(name) =>
      match root.resolve_named(name) {
        Some(kind) =>
          from_json_schema(
            root,
            @schema.Schema::new(kind, name),
            value,
            limits,
            depth + 1,
          )
        None => raise UnresolvedName(name)
      }
  }
}

///|
/// Convert an Avro datum to its schema-defined JSON value.
pub fn to_json(
  schema : @schema.Schema,
  datum : @codec.Datum,
  limits? : JsonLimits = JsonLimits::new(),
) -> Json raise JsonError {
  to_json_schema(schema, schema, datum, limits, 0)
}

///|
/// Convert a JSON value to an Avro datum using the supplied schema.
pub fn from_json(
  schema : @schema.Schema,
  value : Json,
  limits? : JsonLimits = JsonLimits::new(),
) -> @codec.Datum raise JsonError {
  from_json_schema(schema, schema, value, limits, 0)
}

///|
/// Parse JSON text and convert it to an Avro datum using the supplied schema.
#warnings("-deprecated")
pub fn parse(
  schema : @schema.Schema,
  input : String,
  limits? : JsonLimits = JsonLimits::new(),
) -> @codec.Datum raise JsonError {
  check_utf8(limits, input)
  let value = @corejson.parse(input, max_nesting_depth=limits.max_depth()) catch {
    error => raise InvalidJson(error.to_string())
  }
  from_json(schema, value, limits~)
}

///|
/// Convert an Avro datum to compact JSON text according to its schema.
pub fn stringify(
  schema : @schema.Schema,
  datum : @codec.Datum,
  limits? : JsonLimits = JsonLimits::new(),
) -> String raise JsonError {
  to_json(schema, datum, limits~).stringify()
}