lightspeed

//// Data-quality and schema-evolution contracts for ETL pipeline boundaries.

import gleam/int
import gleam/list
import gleam/option.{type Option, None, Some}

/// Boundary field type.
pub type FieldType {
  StringType
  IntType
  BoolType
}

/// One schema field contract.
pub type Field {
  Field(name: String, kind: FieldType, required: Bool)
}

/// One versioned schema boundary.
pub type Schema {
  Schema(name: String, version: Int, fields: List(Field))
}

/// Typed value at one boundary.
pub type Value {
  StringValue(String)
  IntValue(Int)
  BoolValue(Bool)
  NullValue
}

/// Named boundary value.
pub type BoundaryValue {
  BoundaryValue(name: String, value: Value)
}

/// Validation error for one payload/schema pair.
pub type ValidationError {
  InvalidSchema(reason: String)
  MissingField(field: String)
  UnexpectedField(field: String)
  TypeMismatch(field: String, expected: FieldType, actual: String)
}

/// Validation result.
pub type ValidationResult {
  ValidationPassed(normalized: List(BoundaryValue))
  ValidationFailed(errors: List(ValidationError))
}

/// Build one field contract.
pub fn field(name: String, kind: FieldType, required: Bool) -> Field {
  Field(name: name, kind: kind, required: required)
}

/// Build one schema contract.
pub fn schema(name: String, version: Int, fields: List(Field)) -> Schema {
  Schema(name: name, version: version, fields: fields)
}

/// Build one named string value.
pub fn string_value(name: String, value: String) -> BoundaryValue {
  BoundaryValue(name: name, value: StringValue(value))
}

/// Build one named integer value.
pub fn int_value(name: String, value: Int) -> BoundaryValue {
  BoundaryValue(name: name, value: IntValue(value))
}

/// Build one named boolean value.
pub fn bool_value(name: String, value: Bool) -> BoundaryValue {
  BoundaryValue(name: name, value: BoolValue(value))
}

/// Build one named null value.
pub fn null_value(name: String) -> BoundaryValue {
  BoundaryValue(name: name, value: NullValue)
}

/// Validate schema invariants.
pub fn valid_schema(schema: Schema) -> Bool {
  schema.name != ""
  && schema.version > 0
  && schema.fields != []
  && fields_valid(schema.fields)
  && field_names_unique(schema.fields, [])
}

/// Check whether `next` is compatibility-safe relative to `previous`.
pub fn check_compatibility(
  previous: Schema,
  next: Schema,
) -> Result(Nil, String) {
  case previous.name != next.name {
    True -> Error("schema_name_mismatch:" <> previous.name <> ":" <> next.name)
    False ->
      case next.version < previous.version {
        True ->
          Error(
            "schema_version_regression:"
            <> int.to_string(previous.version)
            <> "->"
            <> int.to_string(next.version),
          )
        False ->
          case valid_schema(previous) && valid_schema(next) {
            False -> Error("invalid_schema")
            True ->
              case compatible_previous_fields(previous.fields, next.fields) {
                Error(reason) -> Error(reason)
                Ok(_) -> compatible_new_fields(previous.fields, next.fields)
              }
          }
      }
  }
}

/// Validate one payload against one schema contract.
pub fn validate(
  schema: Schema,
  payload: List(BoundaryValue),
) -> ValidationResult {
  case valid_schema(schema) {
    False -> ValidationFailed(errors: [InvalidSchema(reason: "invalid_schema")])
    True -> {
      let #(normalized_rev, schema_errors_rev) =
        validate_schema_fields(schema.fields, payload, [], [])
      let errors_rev =
        validate_unexpected_payload_fields(
          payload,
          schema.fields,
          schema_errors_rev,
        )

      case errors_rev {
        [] -> ValidationPassed(normalized: list.reverse(normalized_rev))
        _ -> ValidationFailed(errors: list.reverse(errors_rev))
      }
    }
  }
}

/// Field-type label.
pub fn field_type_label(kind: FieldType) -> String {
  case kind {
    StringType -> "string"
    IntType -> "int"
    BoolType -> "bool"
  }
}

/// Value label.
pub fn value_label(value: Value) -> String {
  case value {
    StringValue(v) -> "string:" <> v
    IntValue(v) -> "int:" <> int.to_string(v)
    BoolValue(v) -> "bool:" <> bool_label(v)
    NullValue -> "null"
  }
}

/// Validation-error label.
pub fn validation_error_label(error: ValidationError) -> String {
  case error {
    InvalidSchema(reason) -> "invalid_schema:" <> reason
    MissingField(field) -> "missing_field:" <> field
    UnexpectedField(field) -> "unexpected_field:" <> field
    TypeMismatch(field, expected, actual) ->
      "type_mismatch:"
      <> field
      <> ":expected="
      <> field_type_label(expected)
      <> ":actual="
      <> actual
  }
}

/// Validation-result label.
pub fn validation_result_label(result: ValidationResult) -> String {
  case result {
    ValidationPassed(normalized) ->
      "valid:"
      <> join_with(
        ",",
        list.map(normalized, fn(entry) {
          entry.name <> "=" <> value_label(entry.value)
        }),
      )
    ValidationFailed(errors) ->
      "invalid:" <> join_with(",", list.map(errors, validation_error_label))
  }
}

/// Stable schema signature.
pub fn schema_signature(schema: Schema) -> String {
  "name="
  <> schema.name
  <> "|version="
  <> int.to_string(schema.version)
  <> "|fields="
  <> join_with(",", list.map(schema.fields, field_signature))
}

fn compatible_previous_fields(
  previous_fields: List(Field),
  next_fields: List(Field),
) -> Result(Nil, String) {
  case previous_fields {
    [] -> Ok(Nil)
    [previous, ..rest] ->
      case find_field(next_fields, previous.name) {
        None -> Error("missing_field:" <> previous.name)
        Some(next) ->
          case previous.kind != next.kind {
            True ->
              Error(
                "type_changed:"
                <> previous.name
                <> ":"
                <> field_type_label(previous.kind)
                <> "->"
                <> field_type_label(next.kind),
              )
            False ->
              case previous.required != next.required {
                True -> Error("requiredness_changed:" <> previous.name)
                False -> compatible_previous_fields(rest, next_fields)
              }
          }
      }
  }
}

fn compatible_new_fields(
  previous_fields: List(Field),
  next_fields: List(Field),
) -> Result(Nil, String) {
  case next_fields {
    [] -> Ok(Nil)
    [entry, ..rest] ->
      case find_field(previous_fields, entry.name), entry.required {
        None, True -> Error("new_required_field:" <> entry.name)
        _, _ -> compatible_new_fields(previous_fields, rest)
      }
  }
}

fn validate_schema_fields(
  fields: List(Field),
  payload: List(BoundaryValue),
  normalized_rev: List(BoundaryValue),
  errors_rev: List(ValidationError),
) -> #(List(BoundaryValue), List(ValidationError)) {
  case fields {
    [] -> #(normalized_rev, errors_rev)
    [entry, ..rest] -> {
      let #(next_normalized, next_errors) = case
        find_value(payload, entry.name)
      {
        None ->
          case entry.required {
            True -> #(normalized_rev, [
              MissingField(field: entry.name),
              ..errors_rev
            ])
            False -> #([null_value(entry.name), ..normalized_rev], errors_rev)
          }
        Some(value) ->
          case value_matches(entry, value.value) {
            True -> #([value, ..normalized_rev], errors_rev)
            False -> #(normalized_rev, [
              TypeMismatch(
                field: entry.name,
                expected: entry.kind,
                actual: value_kind_label(value.value),
              ),
              ..errors_rev
            ])
          }
      }

      validate_schema_fields(rest, payload, next_normalized, next_errors)
    }
  }
}

fn validate_unexpected_payload_fields(
  payload: List(BoundaryValue),
  fields: List(Field),
  errors_rev: List(ValidationError),
) -> List(ValidationError) {
  case payload {
    [] -> errors_rev
    [entry, ..rest] ->
      case find_field(fields, entry.name) {
        None ->
          validate_unexpected_payload_fields(rest, fields, [
            UnexpectedField(field: entry.name),
            ..errors_rev
          ])
        Some(_) -> validate_unexpected_payload_fields(rest, fields, errors_rev)
      }
  }
}

fn fields_valid(fields: List(Field)) -> Bool {
  case fields {
    [] -> True
    [entry, ..rest] -> entry.name != "" && fields_valid(rest)
  }
}

fn field_names_unique(fields: List(Field), seen: List(String)) -> Bool {
  case fields {
    [] -> True
    [entry, ..rest] ->
      case contains_name(seen, entry.name) {
        True -> False
        False -> field_names_unique(rest, [entry.name, ..seen])
      }
  }
}

fn find_field(fields: List(Field), name: String) -> Option(Field) {
  case fields {
    [] -> None
    [entry, ..rest] ->
      case entry.name == name {
        True -> Some(entry)
        False -> find_field(rest, name)
      }
  }
}

fn find_value(
  payload: List(BoundaryValue),
  name: String,
) -> Option(BoundaryValue) {
  case payload {
    [] -> None
    [entry, ..rest] ->
      case entry.name == name {
        True -> Some(entry)
        False -> find_value(rest, name)
      }
  }
}

fn value_matches(entry: Field, value: Value) -> Bool {
  case value {
    NullValue -> !entry.required
    StringValue(_) -> entry.kind == StringType
    IntValue(_) -> entry.kind == IntType
    BoolValue(_) -> entry.kind == BoolType
  }
}

fn field_signature(field: Field) -> String {
  field.name
  <> ":"
  <> field_type_label(field.kind)
  <> ":required="
  <> bool_label(field.required)
}

fn value_kind_label(value: Value) -> String {
  case value {
    StringValue(_) -> "string"
    IntValue(_) -> "int"
    BoolValue(_) -> "bool"
    NullValue -> "null"
  }
}

fn contains_name(names: List(String), name: String) -> Bool {
  case names {
    [] -> False
    [entry, ..rest] ->
      case entry == name {
        True -> True
        False -> contains_name(rest, name)
      }
  }
}

fn bool_label(value: Bool) -> String {
  case value {
    True -> "true"
    False -> "false"
  }
}

fn join_with(separator: String, values: List(String)) -> String {
  case values {
    [] -> ""
    [value] -> value
    [value, ..rest] -> value <> separator <> join_with(separator, rest)
  }
}