lightspeed

//// ETL SLO budget contracts for lag/throughput/retry/replay recovery.

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

pub const budget_version = 1

/// One ETL SLO budget profile.
pub type Budget {
  Budget(
    profile: String,
    max_lag_ms: Int,
    min_throughput_records_per_sec: Int,
    max_retry_rate_percent: Int,
    max_replay_recovery_ms: Int,
  )
}

/// One measured ETL observation.
pub type Observation {
  Observation(
    profile: String,
    lag_ms: Int,
    throughput_records_per_sec: Int,
    retry_count: Int,
    processed_records: Int,
    replay_recovery_ms: Int,
  )
}

/// One budget check result.
pub type BudgetResult {
  BudgetResult(profile: String, passed: Bool, reason: String)
}

/// Build one SLO budget profile.
pub fn budget(
  profile: String,
  max_lag_ms: Int,
  min_throughput_records_per_sec: Int,
  max_retry_rate_percent: Int,
  max_replay_recovery_ms: Int,
) -> Budget {
  Budget(
    profile: profile,
    max_lag_ms: max_lag_ms,
    min_throughput_records_per_sec: min_throughput_records_per_sec,
    max_retry_rate_percent: max_retry_rate_percent,
    max_replay_recovery_ms: max_replay_recovery_ms,
  )
}

/// Build one ETL observation.
pub fn observation(
  profile: String,
  lag_ms: Int,
  throughput_records_per_sec: Int,
  retry_count: Int,
  processed_records: Int,
  replay_recovery_ms: Int,
) -> Observation {
  Observation(
    profile: profile,
    lag_ms: lag_ms,
    throughput_records_per_sec: throughput_records_per_sec,
    retry_count: retry_count,
    processed_records: processed_records,
    replay_recovery_ms: replay_recovery_ms,
  )
}

/// Versioned M33 budget profile label.
pub fn budget_version_label() -> String {
  "m33.budget.v" <> int.to_string(budget_version)
}

/// Default M33 ETL SLO budget profile list.
pub fn default_budget() -> List(Budget) {
  [
    Budget(
      profile: "orders_hourly",
      max_lag_ms: 200,
      min_throughput_records_per_sec: 350,
      max_retry_rate_percent: 5,
      max_replay_recovery_ms: 3000,
    ),
    Budget(
      profile: "orders_replay",
      max_lag_ms: 320,
      min_throughput_records_per_sec: 220,
      max_retry_rate_percent: 8,
      max_replay_recovery_ms: 4500,
    ),
  ]
}

/// Validate one budget profile.
pub fn valid_budget(budget: Budget) -> Bool {
  budget.profile != ""
  && budget.max_lag_ms >= 0
  && budget.min_throughput_records_per_sec > 0
  && budget.max_retry_rate_percent >= 0
  && budget.max_replay_recovery_ms >= 0
}

/// Validate one observation.
pub fn valid_observation(observation: Observation) -> Bool {
  observation.profile != ""
  && observation.lag_ms >= 0
  && observation.throughput_records_per_sec >= 0
  && observation.retry_count >= 0
  && observation.processed_records >= 0
  && observation.replay_recovery_ms >= 0
}

/// Validate a budget list and enforce unique profile names.
pub fn valid_budgets(budgets: List(Budget)) -> Bool {
  budgets_valid(budgets) && budget_profiles_unique(budgets, [])
}

/// Evaluate one observation set against one budget list.
pub fn evaluate(
  observations: List(Observation),
  budgets: List(Budget),
) -> List(BudgetResult) {
  evaluate_loop(observations, budgets, [])
}

/// Count budget failures.
pub fn budget_failures(results: List(BudgetResult)) -> Int {
  case results {
    [] -> 0
    [result, ..rest] ->
      case result.passed {
        True -> budget_failures(rest)
        False -> 1 + budget_failures(rest)
      }
  }
}

/// Stable budget-result label.
pub fn budget_result_label(result: BudgetResult) -> String {
  result.profile <> ":" <> bool_label(result.passed) <> ":" <> result.reason
}

/// Stable observation label.
pub fn observation_label(observation: Observation) -> String {
  observation.profile
  <> ":lag_ms="
  <> int.to_string(observation.lag_ms)
  <> ":throughput_rps="
  <> int.to_string(observation.throughput_records_per_sec)
  <> ":retry_rate_percent="
  <> int.to_string(retry_rate_percent(observation))
  <> ":replay_recovery_ms="
  <> int.to_string(observation.replay_recovery_ms)
}

/// Retry-rate percentage derived from retry count and processed records.
pub fn retry_rate_percent(observation: Observation) -> Int {
  case observation.processed_records <= 0 {
    True ->
      case observation.retry_count == 0 {
        True -> 0
        False -> 100
      }
    False ->
      max(
        0,
        { observation.retry_count * 100 }
          / max(1, observation.processed_records),
      )
  }
}

fn evaluate_loop(
  observations: List(Observation),
  budgets: List(Budget),
  results_rev: List(BudgetResult),
) -> List(BudgetResult) {
  case budgets {
    [] -> list.reverse(results_rev)
    [entry, ..rest] ->
      evaluate_loop(observations, rest, [
        evaluate_one(observations, entry),
        ..results_rev
      ])
  }
}

fn evaluate_one(
  observations: List(Observation),
  budget: Budget,
) -> BudgetResult {
  case find_observation(observations, budget.profile) {
    None ->
      BudgetResult(
        profile: budget.profile,
        passed: False,
        reason: "missing_observation",
      )
    Some(observation) ->
      case valid_budget(budget) && valid_observation(observation) {
        False ->
          BudgetResult(
            profile: budget.profile,
            passed: False,
            reason: "invalid_input",
          )
        True -> {
          let lag_ok = observation.lag_ms <= budget.max_lag_ms
          let throughput_ok =
            observation.throughput_records_per_sec
            >= budget.min_throughput_records_per_sec
          let retry_rate_ok =
            retry_rate_percent(observation) <= budget.max_retry_rate_percent
          let replay_recovery_ok =
            observation.replay_recovery_ms <= budget.max_replay_recovery_ms
          let passed =
            lag_ok && throughput_ok && retry_rate_ok && replay_recovery_ok
          let failures =
            failing_checks(
              lag_ok,
              throughput_ok,
              retry_rate_ok,
              replay_recovery_ok,
            )

          BudgetResult(
            profile: budget.profile,
            passed: passed,
            reason: case passed {
              True -> "within_budget"
              False -> join_with("|", failures)
            },
          )
        }
      }
  }
}

fn failing_checks(
  lag_ok: Bool,
  throughput_ok: Bool,
  retry_rate_ok: Bool,
  replay_recovery_ok: Bool,
) -> List(String) {
  []
  |> append_if_false(lag_ok, "lag_exceeded")
  |> append_if_false(throughput_ok, "throughput_below")
  |> append_if_false(retry_rate_ok, "retry_rate_exceeded")
  |> append_if_false(replay_recovery_ok, "replay_recovery_exceeded")
}

fn append_if_false(
  items: List(String),
  condition: Bool,
  value: String,
) -> List(String) {
  case condition {
    True -> items
    False -> list.append(items, [value])
  }
}

fn budgets_valid(budgets: List(Budget)) -> Bool {
  case budgets {
    [] -> True
    [entry, ..rest] -> valid_budget(entry) && budgets_valid(rest)
  }
}

fn budget_profiles_unique(budgets: List(Budget), seen: List(String)) -> Bool {
  case budgets {
    [] -> True
    [entry, ..rest] ->
      case contains_profile(seen, entry.profile) {
        True -> False
        False -> budget_profiles_unique(rest, [entry.profile, ..seen])
      }
  }
}

fn contains_profile(profiles: List(String), profile: String) -> Bool {
  case profiles {
    [] -> False
    [entry, ..rest] ->
      case entry == profile {
        True -> True
        False -> contains_profile(rest, profile)
      }
  }
}

fn find_observation(
  observations: List(Observation),
  profile: String,
) -> Option(Observation) {
  case observations {
    [] -> None
    [entry, ..rest] ->
      case entry.profile == profile {
        True -> Some(entry)
        False -> find_observation(rest, profile)
      }
  }
}

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

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

fn max(left: Int, right: Int) -> Int {
  case left >= right {
    True -> left
    False -> right
  }
}