///|
/// Runtime data contracts for validating records before they reach a model.
///
/// Contracts make the online path explicit: fields are ordered, typed, and
/// checked with the same rules used by batch ingestion.  The implementation
/// deliberately returns structured violations so callers can route bad rows
/// to a dead-letter stream instead of silently dropping them.
pub(all) enum ContractValue {
  Number(Double)
  Text(String)
  Boolean(Bool)
  Vector(Array[Double])
  Missing
} derive(Debug, Eq)

///|
pub(all) enum ContractType {
  Number
  Text
  Boolean
  Vector
} derive(Debug, Eq)

///|
pub(all) enum ContractViolationKind {
  MissingRequired
  TypeMismatch
  BelowMinimum
  AboveMaximum
  InvalidVectorLength
  EmptyText
} derive(Debug, Eq)

///|
pub struct ContractField {
  name : String
  expected : ContractType
  required : Bool
  minimum : Double?
  maximum : Double?
  vector_length : Int
  allow_empty : Bool
}

///|
pub fn ContractField::number(
  name : String,
  required? : Bool = true,
  minimum? : Double,
  maximum? : Double,
) -> ContractField {
  {
    name,
    expected: Number,
    required,
    minimum,
    maximum,
    vector_length: 0,
    allow_empty: true,
  }
}

///|
pub fn ContractField::text(
  name : String,
  required? : Bool = true,
  allow_empty? : Bool = false,
) -> ContractField {
  {
    name,
    expected: Text,
    required,
    minimum: None,
    maximum: None,
    vector_length: 0,
    allow_empty,
  }
}

///|
pub fn ContractField::boolean(
  name : String,
  required? : Bool = true,
) -> ContractField {
  {
    name,
    expected: Boolean,
    required,
    minimum: None,
    maximum: None,
    vector_length: 0,
    allow_empty: true,
  }
}

///|
pub fn ContractField::vector(
  name : String,
  length : Int,
  required? : Bool = true,
) -> ContractField {
  {
    name,
    expected: Vector,
    required,
    minimum: None,
    maximum: None,
    vector_length: if length < 0 {
      0
    } else {
      length
    },
    allow_empty: true,
  }
}

///|
pub fn ContractField::name(self : ContractField) -> String {
  self.name
}

///|
pub fn ContractField::expected(self : ContractField) -> ContractType {
  self.expected
}

///|
pub fn ContractField::required(self : ContractField) -> Bool {
  self.required
}

///|
pub fn ContractField::vector_length(self : ContractField) -> Int {
  self.vector_length
}

///|
pub struct ContractViolation {
  field : String
  kind : ContractViolationKind
  detail : String
}

///|
pub fn ContractViolation::new(
  field : String,
  kind : ContractViolationKind,
  detail : String,
) -> ContractViolation {
  { field, kind, detail }
}

///|
pub fn ContractViolation::field(self : ContractViolation) -> String {
  self.field
}

///|
pub fn ContractViolation::kind(
  self : ContractViolation,
) -> ContractViolationKind {
  self.kind
}

///|
pub fn ContractViolation::detail(self : ContractViolation) -> String {
  self.detail
}

///|
pub struct ContractReport {
  accepted : Bool
  checked_fields : Int
  violations : Array[ContractViolation]
}

///|
pub fn ContractReport::accepted(self : ContractReport) -> Bool {
  self.accepted
}

///|
pub fn ContractReport::checked_fields(self : ContractReport) -> Int {
  self.checked_fields
}

///|
pub fn ContractReport::violations(
  self : ContractReport,
) -> Array[ContractViolation] {
  self.violations.map(value => value)
}

///|
pub fn ContractReport::violation_count(self : ContractReport) -> Int {
  self.violations.length()
}

///|
pub struct DataContract {
  name : String
  version : String
  fields : Array[ContractField]
  mut rows : Int
  mut accepted_rows : Int
  mut rejected_rows : Int
}

///|
pub fn DataContract::new(
  name : String,
  version? : String = "1",
) -> DataContract {
  { name, version, fields: [], rows: 0, accepted_rows: 0, rejected_rows: 0 }
}

///|
pub fn DataContract::name(self : DataContract) -> String {
  self.name
}

///|
pub fn DataContract::version(self : DataContract) -> String {
  self.version
}

///|
pub fn DataContract::add_field(
  self : DataContract,
  field : ContractField,
) -> Bool {
  if self.fields.any(existing => existing.name() == field.name()) {
    false
  } else {
    self.fields.push(field)
    true
  }
}

///|
pub fn DataContract::field_count(self : DataContract) -> Int {
  self.fields.length()
}

///|
pub fn DataContract::fields(self : DataContract) -> Array[ContractField] {
  self.fields.map(value => value)
}

///|
fn contract_type_matches(
  expected : ContractType,
  value : ContractValue,
) -> Bool {
  match (expected, value) {
    (Number, Number(_)) => true
    (Text, Text(_)) => true
    (Boolean, Boolean(_)) => true
    (Vector, Vector(_)) => true
    _ => false
  }
}

///|
fn contract_type_summary(value : ContractType) -> String {
  match value {
    Number => "number"
    Text => "text"
    Boolean => "boolean"
    Vector => "vector"
  }
}

///|
fn contract_value_summary(value : ContractValue) -> String {
  match value {
    Number(_) => "number"
    Text(_) => "text"
    Boolean(_) => "boolean"
    Vector(values) => "vector(\{values.length()})"
    Missing => "missing"
  }
}

///|
fn validate_contract_field(
  field : ContractField,
  value : ContractValue,
  index : Int,
) -> Array[ContractViolation] {
  let violations : Array[ContractViolation] = []
  match value {
    Missing =>
      if field.required {
        violations.push(
          ContractViolation::new(
            field.name,
            MissingRequired,
            "required field is missing at index \{index}",
          ),
        )
      }
    _ =>
      if !contract_type_matches(field.expected, value) {
        violations.push(
          ContractViolation::new(
            field.name,
            TypeMismatch,
            "expected \{contract_type_summary(field.expected)}, received \{contract_value_summary(value)}",
          ),
        )
      } else {
        match value {
          Number(number) => {
            match field.minimum {
              Some(lower) =>
                if number < lower {
                  violations.push(
                    ContractViolation::new(
                      field.name,
                      BelowMinimum,
                      "number is below the configured minimum",
                    ),
                  )
                }
              None => ()
            }
            match field.maximum {
              Some(upper) =>
                if number > upper {
                  violations.push(
                    ContractViolation::new(
                      field.name,
                      AboveMaximum,
                      "number is above the configured maximum",
                    ),
                  )
                }
              None => ()
            }
          }
          Text(text) =>
            if text == "" && !field.allow_empty {
              violations.push(
                ContractViolation::new(
                  field.name,
                  EmptyText,
                  "text fields may not be empty",
                ),
              )
            }
          Vector(values) =>
            if values.length() != field.vector_length {
              violations.push(
                ContractViolation::new(
                  field.name,
                  InvalidVectorLength,
                  "vector length does not match the contract",
                ),
              )
            }
          _ => ()
        }
      }
  }
  violations
}

///|
pub fn DataContract::validate(
  self : DataContract,
  values : Array[ContractValue],
) -> ContractReport {
  self.rows += 1
  let violations : Array[ContractViolation] = []
  for i in 0.. self.fields.length() {
      violations.push(
        ContractViolation::new(
          "",
          TypeMismatch,
          "record contains fields not declared by the contract",
        ),
      )
    }
  }
  { accepted, checked_fields: self.fields.length(), violations }
}

///|
pub fn DataContract::rows(self : DataContract) -> Int {
  self.rows
}

///|
pub fn DataContract::accepted_rows(self : DataContract) -> Int {
  self.accepted_rows
}

///|
pub fn DataContract::rejected_rows(self : DataContract) -> Int {
  self.rejected_rows
}

///|
pub fn DataContract::acceptance_rate(self : DataContract) -> Double {
  if self.rows == 0 {
    0.0
  } else {
    self.accepted_rows.to_double() / self.rows.to_double()
  }
}

///|
pub fn DataContract::reset(self : DataContract) -> Unit {
  self.rows = 0
  self.accepted_rows = 0
  self.rejected_rows = 0
}

///|
/// Bounded idempotency ledger for event consumers.  A key is accepted only
/// once while it remains in the ledger; evictions are deterministic FIFO.
pub struct IdempotencyLedger {
  capacity : Int
  keys : Map[String, Int64]
  order : Array[String]
  mut duplicates : Int
  mut evictions : Int
}

///|
pub fn IdempotencyLedger::new(capacity? : Int = 4096) -> IdempotencyLedger {
  {
    capacity: if capacity < 0 {
      0
    } else {
      capacity
    },
    keys: {},
    order: [],
    duplicates: 0,
    evictions: 0,
  }
}

///|
pub fn IdempotencyLedger::accept(
  self : IdempotencyLedger,
  key : String,
  timestamp : Int64,
) -> Bool {
  if self.keys.contains(key) {
    self.duplicates += 1
    false
  } else if self.capacity == 0 {
    true
  } else {
    self.keys[key] = timestamp
    self.order.push(key)
    if self.order.length() > self.capacity {
      let oldest = self.order.remove(0)
      self.keys.remove(oldest)
      self.evictions += 1
    }
    true
  }
}

///|
pub fn IdempotencyLedger::contains(
  self : IdempotencyLedger,
  key : String,
) -> Bool {
  self.keys.contains(key)
}

///|
pub fn IdempotencyLedger::timestamp(
  self : IdempotencyLedger,
  key : String,
) -> Int64? {
  self.keys.get(key)
}

///|
pub fn IdempotencyLedger::size(self : IdempotencyLedger) -> Int {
  self.keys.length()
}

///|
pub fn IdempotencyLedger::duplicates(self : IdempotencyLedger) -> Int {
  self.duplicates
}

///|
pub fn IdempotencyLedger::evictions(self : IdempotencyLedger) -> Int {
  self.evictions
}

///|
pub fn IdempotencyLedger::reset(self : IdempotencyLedger) -> Unit {
  self.keys.clear()
  self.order.clear()
  self.duplicates = 0
  self.evictions = 0
}

///|
pub struct RowQualitySummary {
  mut rows : Int
  mut accepted : Int
  mut rejected : Int
  mut duplicates : Int
  mut missing_values : Int
  mut type_errors : Int
}

///|
pub fn RowQualitySummary::new() -> RowQualitySummary {
  {
    rows: 0,
    accepted: 0,
    rejected: 0,
    duplicates: 0,
    missing_values: 0,
    type_errors: 0,
  }
}

///|
pub fn RowQualitySummary::observe(
  self : RowQualitySummary,
  report : ContractReport,
  duplicate : Bool,
) -> Unit {
  self.rows += 1
  if duplicate {
    self.duplicates += 1
  }
  if report.accepted() && !duplicate {
    self.accepted += 1
  } else {
    self.rejected += 1
  }
  for violation in report.violations() {
    match violation.kind() {
      MissingRequired => self.missing_values += 1
      TypeMismatch => self.type_errors += 1
      _ => ()
    }
  }
}

///|
pub fn RowQualitySummary::rows(self : RowQualitySummary) -> Int {
  self.rows
}

///|
pub fn RowQualitySummary::accepted(self : RowQualitySummary) -> Int {
  self.accepted
}

///|
pub fn RowQualitySummary::rejected(self : RowQualitySummary) -> Int {
  self.rejected
}

///|
pub fn RowQualitySummary::duplicates(self : RowQualitySummary) -> Int {
  self.duplicates
}

///|
pub fn RowQualitySummary::missing_values(self : RowQualitySummary) -> Int {
  self.missing_values
}

///|
pub fn RowQualitySummary::type_errors(self : RowQualitySummary) -> Int {
  self.type_errors
}

///|
pub fn RowQualitySummary::acceptance_rate(self : RowQualitySummary) -> Double {
  if self.rows == 0 {
    0.0
  } else {
    self.accepted.to_double() / self.rows.to_double()
  }
}

///|
pub fn RowQualitySummary::reset(self : RowQualitySummary) -> Unit {
  self.rows = 0
  self.accepted = 0
  self.rejected = 0
  self.duplicates = 0
  self.missing_values = 0
  self.type_errors = 0
}

///|
/// Combines contract validation and idempotency checks into one ingestion
/// gate.  It is intentionally stateful so its counters can feed monitoring.
pub struct IngestionGate {
  contract : DataContract
  ledger : IdempotencyLedger
  summary : RowQualitySummary
}

///|
pub fn IngestionGate::new(
  contract : DataContract,
  ledger_capacity? : Int = 4096,
) -> IngestionGate {
  {
    contract,
    ledger: IdempotencyLedger::new(capacity=ledger_capacity),
    summary: RowQualitySummary::new(),
  }
}

///|
pub fn IngestionGate::accept(
  self : IngestionGate,
  key : String,
  timestamp : Int64,
  values : Array[ContractValue],
) -> Bool {
  let report = self.contract.validate(values)
  let fresh = self.ledger.accept(key, timestamp)
  self.summary.observe(report, !fresh)
  fresh && report.accepted()
}

///|
pub fn IngestionGate::contract(self : IngestionGate) -> DataContract {
  self.contract
}

///|
pub fn IngestionGate::summary(self : IngestionGate) -> RowQualitySummary {
  self.summary
}

///|
pub fn IngestionGate::ledger(self : IngestionGate) -> IdempotencyLedger {
  self.ledger
}

///|
pub fn IngestionGate::reset(self : IngestionGate) -> Unit {
  self.contract.reset()
  self.ledger.reset()
  self.summary.reset()
}