///|
fn schema_change(
  changes : Array[Value],
  path : String,
  classification : String,
  reason : String,
) -> Unit {
  changes.push(
    record([
      ("path", String(path)),
      ("classification", String(classification)),
      ("reason", String(reason)),
    ]),
  )
}

///|
fn compare_schema(
  old_doc : Value,
  new_doc : Value,
  old : Value,
  next : Value,
  direction : String,
  path : String,
  changes : Array[Value],
  depth : Int32,
) -> Unit raise {
  if depth > 32 {
    schema_change(changes, path, "review", "reference or schema depth limit")
    return
  }
  if get(old, "$ref") != Null {
    compare_schema(
      old_doc,
      new_doc,
      pointer(old_doc, str(get(old, "$ref"))),
      next,
      direction,
      path,
      changes,
      depth + 1,
    )
    return
  }
  if get(next, "$ref") != Null {
    compare_schema(
      old_doc,
      new_doc,
      old,
      pointer(new_doc, str(get(next, "$ref"))),
      direction,
      path,
      changes,
      depth + 1,
    )
    return
  }
  if old == next {
    return
  }
  ignore(obj(old))
  ignore(obj(next))
  let tightening = direction == "request"
  let known = [
    "type", "nullable", "enum", "required", "properties", "minimum", "maximum", "minLength",
    "maxLength", "minItems", "maxItems", "items", "title", "description", "example",
    "default", "deprecated",
  ]
  let keys : Map[String, Bool] = Map([])
  for key, _ in obj(old) {
    keys[key] = true
  }
  for key, _ in obj(next) {
    keys[key] = true
  }
  for key, _ in keys {
    if !known.contains(key) {
      schema_change(
        changes,
        path + "/" + key,
        "review",
        "unsupported constraint participates in changed schema",
      )
    }
  }
  let a = get(old, "type")
  let b = get(next, "type")
  if a != b {
    let narrows = (a == String("number") && b == String("integer")) ||
      (a == Null && b != Null)
    let widens = (a == String("integer") && b == String("number")) ||
      (a != Null && b == Null)
    schema_change(
      changes,
      path + "/type",
      if (tightening && narrows) ||
        (!tightening && widens) ||
        (!narrows && !widens) {
        "breaking"
      } else {
        "compatible"
      },
      "type domain changed",
    )
  }
  let old_nullable = get(old, "nullable") == Bool(true)
  let new_nullable = get(next, "nullable") == Bool(true)
  if old_nullable != new_nullable {
    schema_change(
      changes,
      path + "/nullable",
      if (tightening && !new_nullable) || (!tightening && new_nullable) {
        "breaking"
      } else {
        "compatible"
      },
      "null domain changed",
    )
  }
  if get(old, "enum") != get(next, "enum") {
    let left = get(old, "enum")
    let right = get(next, "enum")
    let removed = left == Null ||
      (
        right != Null &&
        arr(left).any(fn(value) { !arr(right).contains(value) })
      )
    let added = right == Null ||
      (left != Null && arr(right).any(fn(value) { !arr(left).contains(value) }))
    schema_change(
      changes,
      path + "/enum",
      if (tightening && removed) || (!tightening && added) {
        "breaking"
      } else {
        "compatible"
      },
      "enum domain changed",
    )
  }
  for
    key in [
      "minimum", "maximum", "minLength", "maxLength", "minItems", "maxItems",
    ] {
    let x = get(old, key)
    let y = get(next, key)
    if x == y {
      continue
    }
    let lower = key.has_prefix("min")
    let narrows = x == Null ||
      (y != Null && (if lower { num(y) > num(x) } else { num(y) < num(x) }))
    schema_change(
      changes,
      path + "/" + key,
      if (tightening && narrows) || (!tightening && !narrows) {
        "breaking"
      } else {
        "compatible"
      },
      "range changed",
    )
  }
  let old_required = if get(old, "required") == Null {
    []
  } else {
    string_list(get(old, "required"))
  }
  let new_required = if get(next, "required") == Null {
    []
  } else {
    string_list(get(next, "required"))
  }
  for field in new_required {
    if !old_required.contains(field) {
      schema_change(
        changes,
        path + "/required/" + field,
        if tightening {
          "breaking"
        } else {
          "compatible"
        },
        "required property added",
      )
    }
  }
  for field in old_required {
    if !new_required.contains(field) {
      schema_change(
        changes,
        path + "/required/" + field,
        if tightening {
          "compatible"
        } else {
          "breaking"
        },
        "required property removed",
      )
    }
  }
  let left = if get(old, "properties") == Null {
    Map([])
  } else {
    obj(get(old, "properties"))
  }
  let right = if get(next, "properties") == Null {
    Map([])
  } else {
    obj(get(next, "properties"))
  }
  for field, schema in left {
    match right.get(field) {
      Some(other) =>
        compare_schema(
          old_doc,
          new_doc,
          schema,
          other,
          direction,
          path + "/properties/" + field,
          changes,
          depth + 1,
        )
      None =>
        schema_change(
          changes,
          path + "/properties/" + field,
          "review",
          "property declaration removed; additionalProperties and consumers require review",
        )
    }
  }
  for field, _ in right {
    if !left.contains(field) {
      schema_change(
        changes,
        path + "/properties/" + field,
        "review",
        "property declaration added; assess optionality and consumer handling",
      )
    }
  }
  if get(old, "items") != Null && get(next, "items") != Null {
    compare_schema(
      old_doc,
      new_doc,
      get(old, "items"),
      get(next, "items"),
      direction,
      path + "/items",
      changes,
      depth + 1,
    )
  } else if get(old, "items") != get(next, "items") {
    schema_change(
      changes,
      path + "/items",
      "review",
      "array item schema added or removed",
    )
  }
}

///|
pub fn schema_compatibility(request : Value) -> Value raise {
  let direction = str(get(request, "direction"))
  if direction != "request" && direction != "response" {
    raise InputError("direction must be request or response")
  }
  let changes : Array[Value] = []
  compare_schema(
    get(request, "old_document"),
    get(request, "new_document"),
    get(request, "old"),
    get(request, "new"),
    direction,
    "",
    changes,
    0,
  )
  record([
    ("direction", String(direction)),
    ("changes", Array(changes)),
    (
      "breaking",
      Bool(
        changes.any(fn(c) { get(c, "classification") == String("breaking") }),
      ),
    ),
    (
      "needs_review",
      Bool(changes.any(fn(c) { get(c, "classification") == String("review") })),
    ),
    (
      "scope",
      String(
        "directional structural schema profile; not full logical implication",
      ),
    ),
  ])
}