///|
/// Errors produced while parsing, naming, or validating an Avro schema.
pub(all) suberror SchemaError {
  EmptySchema
  UnsupportedSchema(String)
  DuplicateName(String)
  UnknownName(String)
  InvalidJson(String)
  InvalidSchema(String)
  InvalidName(String)
  InvalidDefault(field~ : String, reason~ : String)
} derive(Debug, Eq)

///|
fn primitive_from_name(name : String) -> Primitive? {
  match name {
    "null" => Some(Null)
    "boolean" => Some(Boolean)
    "int" => Some(Int)
    "long" => Some(Long)
    "float" => Some(Float)
    "double" => Some(Double)
    "bytes" => Some(Bytes)
    "string" => Some(String)
    _ => None
  }
}

///|
fn is_name_start(character : Char) -> Bool {
  character == '_' || character is ('a'..='z') || character is ('A'..='Z')
}

///|
fn is_name_character(character : Char) -> Bool {
  is_name_start(character) || character is ('0'..='9')
}

///|
fn validate_simple_name(name : String) -> Unit raise SchemaError {
  let mut first = true
  for character in name {
    if (first && !is_name_start(character)) ||
      (!first && !is_name_character(character)) {
      raise InvalidName(name)
    }
    first = false
  }
  if first {
    raise InvalidName(name)
  }
}

///|
fn validate_full_name(name : String) -> Unit raise SchemaError {
  for part in name.split(".") {
    validate_simple_name(part.to_owned())
  }
}

///|
fn full_name(name : String, scope : String?) -> String raise SchemaError {
  if name.contains(".") {
    validate_full_name(name)
    name
  } else {
    validate_simple_name(name)
    match scope {
      None => name
      Some(value) => {
        validate_full_name(value)
        "\{value}.\{name}"
      }
    }
  }
}

///|
fn namespace_of(name : String) -> String? {
  let parts = name.split(".").map(part => part.to_owned()).to_array()
  if parts.length() <= 1 {
    None
  } else {
    Some(parts[:parts.length() - 1].join("."))
  }
}

///|
fn required(object : Map[String, Json], key : String) -> Json raise SchemaError {
  match object.get(key) {
    Some(value) => value
    None => raise InvalidSchema("missing required property '\{key}'")
  }
}

///|
fn required_string(
  object : Map[String, Json],
  key : String,
) -> String raise SchemaError {
  match required(object, key) {
    String(value) => value
    _ => raise InvalidSchema("property '\{key}' must be a string")
  }
}

///|
fn optional_string(
  object : Map[String, Json],
  key : String,
) -> String? raise SchemaError {
  match object.get(key) {
    None | Some(Null) => None
    Some(String(value)) => Some(value)
    Some(_) => raise InvalidSchema("property '\{key}' must be a string")
  }
}

///|
fn parse_string_array(
  value : Json,
  property : String,
) -> Array[String] raise SchemaError {
  match value {
    Array(values) =>
      values.map(item => {
        match item {
          String(text) => text
          _ =>
            raise InvalidSchema("property '\{property}' must contain strings")
        }
      })
    _ => raise InvalidSchema("property '\{property}' must be an array")
  }
}

///|
fn aliases(
  object : Map[String, Json],
  scope : String?,
) -> Array[String] raise SchemaError {
  match object.get("aliases") {
    None => []
    Some(value) =>
      parse_string_array(value, "aliases").map(alternate => {
        full_name(alternate, scope)
      })
  }
}

///|
fn field_aliases(object : Map[String, Json]) -> Array[String] raise SchemaError {
  match object.get("aliases") {
    None => []
    Some(value) => {
      let values = parse_string_array(value, "aliases")
      for alternate in values {
        validate_simple_name(alternate)
      }
      values
    }
  }
}

///|
fn register_name(
  names : Map[String, String],
  name : String,
  type_aliases : Array[String],
) -> Unit raise SchemaError {
  if names.contains(name) {
    raise DuplicateName(name)
  }
  names[name] = name
  for alternate in type_aliases {
    if names.contains(alternate) {
      raise DuplicateName(alternate)
    }
    names[alternate] = name
  }
}

///|
fn resolve_reference(
  name : String,
  scope : String?,
  names : Map[String, String],
) -> SchemaKind raise SchemaError {
  match primitive_from_name(name) {
    Some(primitive) => Primitive(primitive)
    None => {
      let candidate = full_name(name, scope)
      match names.get(candidate) {
        Some(full) => Named(full)
        None => raise UnknownName(candidate)
      }
    }
  }
}

///|
fn definition_key(kind : SchemaKind) -> String {
  match kind {
    Primitive(Null) => "primitive:null"
    Primitive(Boolean) => "primitive:boolean"
    Primitive(Int) => "primitive:int"
    Primitive(Long) => "primitive:long"
    Primitive(Float) => "primitive:float"
    Primitive(Double) => "primitive:double"
    Primitive(Bytes) => "primitive:bytes"
    Primitive(String) => "primitive:string"
    Record(name~, namespace_=_, aliases=_, fields=_) => "record:\{name}"
    Enum(name~, namespace_=_, aliases=_, symbols=_, default_symbol=_) =>
      "enum:\{name}"
    Fixed(name~, namespace_=_, aliases=_, size=_) => "fixed:\{name}"
    Named(name) => "named:\{name}"
    Array(items=_) => "array"
    Map(values=_) => "map"
    Union(branches=_) => "union"
  }
}

///|
fn valid_default(schema : Schema, value : Json) -> Bool {
  match schema.kind {
    Primitive(Null) => value is Null
    Primitive(Boolean) => value is (True | False)
    Primitive(Int) =>
      match value {
        Number(number, repr=_) => number == number.to_int().to_double()
        _ => false
      }
    Primitive(Long) =>
      match value {
        Number(number, repr=_) => number == number.to_int().to_double()
        _ => false
      }
    Primitive(Float) | Primitive(Double) => value is Number(_, repr=_)
    Primitive(Bytes) | Primitive(String) => value is String(_)
    Array(items~) =>
      match value {
        Array(values) => values.all(item => valid_default(items, item))
        _ => false
      }
    Map(values~) =>
      match value {
        Object(entries) =>
          entries.values().all(item => valid_default(values, item))
        _ => false
      }
    Union(branches~) =>
      match branches.get(0) {
        Some(first) => valid_default(first, value)
        None => false
      }
    Enum(symbols~, ..) =>
      match value {
        String(symbol) => symbols.contains(symbol)
        _ => false
      }
    Fixed(size~, ..) =>
      match value {
        String(text) => text.length() == size
        _ => false
      }
    Record(fields~, ..) =>
      match value {
        Object(object) =>
          fields.all(field => {
            match object.get(field.name) {
              Some(field_value) => valid_default(field.schema, field_value)
              None => field.default_value is Some(_)
            }
          })
        _ => false
      }
    Named(_) => true
  }
}

///|
fn make_schema(kind : SchemaKind) -> Schema {
  { kind, source: "", definitions: [] }
}

///|
fn parse_field(
  value : Json,
  scope : String?,
  names : Map[String, String],
  definitions : Array[NamedDefinition],
) -> Field raise SchemaError {
  let object = match value {
    Object(entries) => entries
    _ => raise InvalidSchema("record fields must be objects")
  }
  let name = required_string(object, "name")
  validate_simple_name(name)
  let schema = parse_value(required(object, "type"), scope, names, definitions)
  let default_value = object.get("default")
  match default_value {
    Some(default) =>
      if !valid_default(schema, default) {
        raise InvalidDefault(
          field=name,
          reason="default does not match the field schema",
        )
      }
    None => ()
  }
  {
    name,
    schema,
    default_value,
    aliases: field_aliases(object),
    doc: optional_string(object, "doc"),
  }
}

///|
fn parse_record(
  object : Map[String, Json],
  inherited_namespace : String?,
  names : Map[String, String],
  definitions : Array[NamedDefinition],
) -> SchemaKind raise SchemaError {
  let selected_namespace = match optional_string(object, "namespace") {
    Some(value) => Some(value)
    None => inherited_namespace
  }
  let full = full_name(required_string(object, "name"), selected_namespace)
  let scope = namespace_of(full)
  let type_aliases = aliases(object, scope)
  register_name(names, full, type_aliases)
  let field_values = match required(object, "fields") {
    Array(values) => values
    _ => raise InvalidSchema("record property 'fields' must be an array")
  }
  let fields = field_values.map(field => {
    parse_field(field, scope, names, definitions)
  })
  let kind = Record(name=full, namespace_=scope, aliases=type_aliases, fields~)
  definitions.push({ name: full, kind })
  kind
}

///|
fn parse_enum(
  object : Map[String, Json],
  inherited_namespace : String?,
  names : Map[String, String],
  definitions : Array[NamedDefinition],
) -> SchemaKind raise SchemaError {
  let selected_namespace = match optional_string(object, "namespace") {
    Some(value) => Some(value)
    None => inherited_namespace
  }
  let full = full_name(required_string(object, "name"), selected_namespace)
  let scope = namespace_of(full)
  let type_aliases = aliases(object, scope)
  register_name(names, full, type_aliases)
  let symbols = parse_string_array(required(object, "symbols"), "symbols")
  if symbols.is_empty() {
    raise InvalidSchema("enum must contain at least one symbol")
  }
  let seen : Map[String, Unit] = Map([])
  for symbol in symbols {
    validate_simple_name(symbol)
    if seen.contains(symbol) {
      raise InvalidSchema("duplicate enum symbol '\{symbol}'")
    }
    seen[symbol] = ()
  }
  let default_symbol = optional_string(object, "default")
  match default_symbol {
    Some(symbol) if !symbols.contains(symbol) =>
      raise InvalidSchema("enum default is not a symbol")
    _ => ()
  }
  let kind = Enum(
    name=full,
    namespace_=scope,
    aliases=type_aliases,
    symbols~,
    default_symbol~,
  )
  definitions.push({ name: full, kind })
  kind
}

///|
fn parse_fixed(
  object : Map[String, Json],
  inherited_namespace : String?,
  names : Map[String, String],
  definitions : Array[NamedDefinition],
) -> SchemaKind raise SchemaError {
  let selected_namespace = match optional_string(object, "namespace") {
    Some(value) => Some(value)
    None => inherited_namespace
  }
  let full = full_name(required_string(object, "name"), selected_namespace)
  let scope = namespace_of(full)
  let type_aliases = aliases(object, scope)
  register_name(names, full, type_aliases)
  let size = match required(object, "size") {
    Number(number, repr=_) if number == number.to_int().to_double() &&
      number.to_int() > 0 => number.to_int()
    _ => raise InvalidSchema("fixed property 'size' must be a positive integer")
  }
  let kind = Fixed(name=full, namespace_=scope, aliases=type_aliases, size~)
  definitions.push({ name: full, kind })
  kind
}

///|
fn parse_union(
  values : Array[Json],
  scope : String?,
  names : Map[String, String],
  definitions : Array[NamedDefinition],
) -> SchemaKind raise SchemaError {
  if values.is_empty() {
    raise InvalidSchema("union must contain at least one branch")
  }
  let branches = values.map(value => {
    parse_value(value, scope, names, definitions)
  })
  let seen : Map[String, Unit] = Map([])
  for branch in branches {
    match branch.kind {
      Union(branches=_) =>
        raise InvalidSchema("union may not contain another union directly")
      _ => ()
    }
    let key = definition_key(branch.kind)
    if seen.contains(key) {
      raise InvalidSchema("union contains duplicate branch '\{key}'")
    }
    seen[key] = ()
  }
  Union(branches~)
}

///|
fn parse_object(
  object : Map[String, Json],
  scope : String?,
  names : Map[String, String],
  definitions : Array[NamedDefinition],
) -> SchemaKind raise SchemaError {
  match required(object, "type") {
    String("record") => parse_record(object, scope, names, definitions)
    String("enum") => parse_enum(object, scope, names, definitions)
    String("fixed") => parse_fixed(object, scope, names, definitions)
    String("array") => {
      let items = parse_value(
        required(object, "items"),
        scope,
        names,
        definitions,
      )
      Array(items~)
    }
    String("map") => {
      let values = parse_value(
        required(object, "values"),
        scope,
        names,
        definitions,
      )
      Map(values~)
    }
    String(type_name) => resolve_reference(type_name, scope, names)
    Array(values) => parse_union(values, scope, names, definitions)
    Object(nested) => parse_object(nested, scope, names, definitions)
    _ => raise InvalidSchema("property 'type' must be a schema")
  }
}

///|
fn parse_value(
  value : Json,
  scope : String?,
  names : Map[String, String],
  definitions : Array[NamedDefinition],
) -> Schema raise SchemaError {
  let kind = match value {
    String(name) => resolve_reference(name, scope, names)
    Array(values) => parse_union(values, scope, names, definitions)
    Object(object) => parse_object(object, scope, names, definitions)
    _ => raise InvalidSchema("a schema must be a string, object, or array")
  }
  make_schema(kind)
}

///|
/// Parse an Avro JSON schema, resolve named references, and retain named
/// declarations for later datum and OCF consumers.
pub fn parse(input : String) -> Schema raise SchemaError {
  let source = input.trim().to_owned()
  if source.is_empty() {
    raise EmptySchema
  }
  let json = @json.parse(source) catch {
    _ => raise InvalidJson("invalid JSON")
  }
  let names : Map[String, String] = Map([])
  let definitions : Array[NamedDefinition] = []
  let parsed = parse_value(json, None, names, definitions)
  { kind: parsed.kind, source, definitions }
}