lightspeed

//// Async data engine and enterprise driver-layer contracts for M48.

import gleam/int
import gleam/list
import gleam/string

/// Enterprise driver boundary class.
pub type DriverClass {
  RelationalAdapter
  ColumnarAdapter
  StreamAdapter
  ObjectStoreAdapter
}

/// Driver capability contract.
pub type DriverCapability {
  NonBlockingExecute
  BoundedConcurrency
  TimeoutTelemetry
  RetryTelemetry
  BatchRead
  BatchWrite
  CheckpointResume
  IdempotentWrite
  MixedLatencyTolerance
}

/// Timeout policy for one driver.
pub type TimeoutPolicy {
  TimeoutPolicy(connect_ms: Int, request_ms: Int, idle_ms: Int)
}

/// Retry policy for one driver.
pub type RetryPolicy {
  RetryPolicy(
    max_attempts: Int,
    base_backoff_ms: Int,
    max_backoff_ms: Int,
    jitter_percent: Int,
  )
}

/// Timeout/retry/in-flight telemetry profile.
pub type TelemetryProfile {
  TelemetryProfile(
    timeout_metric: String,
    retry_metric: String,
    latency_metric: String,
    in_flight_metric: String,
  )
}

/// One enterprise async driver contract.
pub type DriverContract {
  DriverContract(
    name: String,
    class: DriverClass,
    provider_module: String,
    max_in_flight: Int,
    max_queue_depth: Int,
    timeout_policy: TimeoutPolicy,
    retry_policy: RetryPolicy,
    telemetry: TelemetryProfile,
    capabilities: List(DriverCapability),
    support_tier: String,
    tuning_guide: String,
  )
}

/// Benchmark profile class for M48 coverage.
pub type BenchmarkProfile {
  HeavyData
  MixedLatency
}

/// Deterministic benchmark sample.
pub type Benchmark {
  Benchmark(
    driver: String,
    profile: BenchmarkProfile,
    p50_latency_ms: Int,
    p95_latency_ms: Int,
    timeout_rate_bps: Int,
    retry_rate_bps: Int,
    throughput_ops_per_sec: Int,
  )
}

/// Budget for one benchmark profile.
pub type Budget {
  Budget(
    profile: BenchmarkProfile,
    max_p95_latency_ms: Int,
    max_timeout_rate_bps: Int,
    max_retry_rate_bps: Int,
    min_throughput_ops_per_sec: Int,
  )
}

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

/// Relational async driver.
pub fn postgres_async_driver() -> DriverContract {
  DriverContract(
    name: "postgres_async_driver",
    class: RelationalAdapter,
    provider_module: "Lightspeed.Driver.PostgresAsync",
    max_in_flight: 32,
    max_queue_depth: 512,
    timeout_policy: timeout_policy(80, 450, 3000),
    retry_policy: retry_policy(4, 25, 250, 15),
    telemetry: telemetry_profile("postgres_async_driver"),
    capabilities: [
      NonBlockingExecute,
      BoundedConcurrency,
      TimeoutTelemetry,
      RetryTelemetry,
      BatchRead,
      BatchWrite,
      IdempotentWrite,
      MixedLatencyTolerance,
    ],
    support_tier: "ga",
    tuning_guide: "docs/async_data_engine_enterprise_driver_expansion.md#postgres-async-driver",
  )
}

/// Columnar analytics async driver.
pub fn clickhouse_async_driver() -> DriverContract {
  DriverContract(
    name: "clickhouse_async_driver",
    class: ColumnarAdapter,
    provider_module: "Lightspeed.Driver.ClickHouseAsync",
    max_in_flight: 24,
    max_queue_depth: 384,
    timeout_policy: timeout_policy(100, 550, 3200),
    retry_policy: retry_policy(4, 30, 280, 15),
    telemetry: telemetry_profile("clickhouse_async_driver"),
    capabilities: [
      NonBlockingExecute,
      BoundedConcurrency,
      TimeoutTelemetry,
      RetryTelemetry,
      BatchRead,
      CheckpointResume,
      MixedLatencyTolerance,
    ],
    support_tier: "ga",
    tuning_guide: "docs/async_data_engine_enterprise_driver_expansion.md#clickhouse-async-driver",
  )
}

/// Streaming async driver.
pub fn kafka_async_driver() -> DriverContract {
  DriverContract(
    name: "kafka_async_driver",
    class: StreamAdapter,
    provider_module: "Lightspeed.Driver.KafkaAsync",
    max_in_flight: 48,
    max_queue_depth: 640,
    timeout_policy: timeout_policy(70, 420, 2800),
    retry_policy: retry_policy(5, 20, 220, 12),
    telemetry: telemetry_profile("kafka_async_driver"),
    capabilities: [
      NonBlockingExecute,
      BoundedConcurrency,
      TimeoutTelemetry,
      RetryTelemetry,
      BatchRead,
      BatchWrite,
      CheckpointResume,
      IdempotentWrite,
      MixedLatencyTolerance,
    ],
    support_tier: "ga",
    tuning_guide: "docs/async_data_engine_enterprise_driver_expansion.md#kafka-async-driver",
  )
}

/// Object-store batch async driver.
pub fn object_store_async_driver() -> DriverContract {
  DriverContract(
    name: "object_store_async_driver",
    class: ObjectStoreAdapter,
    provider_module: "Lightspeed.Driver.ObjectStoreAsync",
    max_in_flight: 20,
    max_queue_depth: 300,
    timeout_policy: timeout_policy(120, 700, 4500),
    retry_policy: retry_policy(4, 35, 320, 20),
    telemetry: telemetry_profile("object_store_async_driver"),
    capabilities: [
      NonBlockingExecute,
      BoundedConcurrency,
      TimeoutTelemetry,
      RetryTelemetry,
      BatchRead,
      BatchWrite,
      CheckpointResume,
      MixedLatencyTolerance,
    ],
    support_tier: "ga",
    tuning_guide: "docs/async_data_engine_enterprise_driver_expansion.md#object-store-async-driver",
  )
}

/// Supported M48 enterprise drivers.
pub fn supported_drivers() -> List(DriverContract) {
  [
    postgres_async_driver(),
    clickhouse_async_driver(),
    kafka_async_driver(),
    object_store_async_driver(),
  ]
}

/// M48 default benchmark budgets.
pub fn default_budgets() -> List(Budget) {
  [
    Budget(
      profile: HeavyData,
      max_p95_latency_ms: 140,
      max_timeout_rate_bps: 24,
      max_retry_rate_bps: 48,
      min_throughput_ops_per_sec: 5000,
    ),
    Budget(
      profile: MixedLatency,
      max_p95_latency_ms: 130,
      max_timeout_rate_bps: 26,
      max_retry_rate_bps: 52,
      min_throughput_ops_per_sec: 5400,
    ),
  ]
}

/// Validate one driver contract.
pub fn valid(driver: DriverContract) -> Bool {
  driver.name != ""
  && driver.provider_module != ""
  && driver.support_tier == "ga"
  && driver.tuning_guide != ""
  && driver.max_in_flight > 0
  && driver.max_queue_depth >= driver.max_in_flight
  && timeout_policy_valid(driver.timeout_policy)
  && retry_policy_valid(driver.retry_policy)
  && has_required_capabilities(driver.capabilities)
  && telemetry_profile_valid(driver.telemetry, driver.name)
}

/// True when all supported drivers are valid and policy-consistent.
pub fn contracts_consistent() -> Bool {
  all_drivers_valid(supported_drivers())
  && timeout_retry_ranges_consistent(supported_drivers())
  && timeout_retry_telemetry_parity()
}

/// True when timeout/retry telemetry contracts are consistent across drivers.
pub fn timeout_retry_telemetry_parity() -> Bool {
  all_telemetry_metrics_present(supported_drivers())
  && telemetry_metric_suffixes_consistent(supported_drivers())
}

/// Stable class label.
pub fn class_label(class: DriverClass) -> String {
  case class {
    RelationalAdapter -> "relational_adapter"
    ColumnarAdapter -> "columnar_adapter"
    StreamAdapter -> "stream_adapter"
    ObjectStoreAdapter -> "object_store_adapter"
  }
}

/// Stable capability label.
pub fn capability_label(capability: DriverCapability) -> String {
  case capability {
    NonBlockingExecute -> "non_blocking_execute"
    BoundedConcurrency -> "bounded_concurrency"
    TimeoutTelemetry -> "timeout_telemetry"
    RetryTelemetry -> "retry_telemetry"
    BatchRead -> "batch_read"
    BatchWrite -> "batch_write"
    CheckpointResume -> "checkpoint_resume"
    IdempotentWrite -> "idempotent_write"
    MixedLatencyTolerance -> "mixed_latency_tolerance"
  }
}

/// Stable benchmark-profile label.
pub fn profile_label(profile: BenchmarkProfile) -> String {
  case profile {
    HeavyData -> "heavy_data"
    MixedLatency -> "mixed_latency"
  }
}

/// Stable driver contract signature.
pub fn driver_signature(driver: DriverContract) -> String {
  "driver="
  <> driver.name
  <> "|class="
  <> class_label(driver.class)
  <> "|provider="
  <> driver.provider_module
  <> "|in_flight="
  <> int.to_string(driver.max_in_flight)
  <> "|queue_depth="
  <> int.to_string(driver.max_queue_depth)
  <> "|timeouts="
  <> timeout_policy_signature(driver.timeout_policy)
  <> "|retries="
  <> retry_policy_signature(driver.retry_policy)
  <> "|telemetry="
  <> telemetry_signature(driver.telemetry)
  <> "|capabilities="
  <> join_with(",", list.map(driver.capabilities, capability_label))
  <> "|tier="
  <> driver.support_tier
}

/// Stable adapter capability matrix signature.
pub fn capability_matrix_signature() -> String {
  join_with(";", list.map(supported_drivers(), driver_signature))
}

/// Run deterministic heavy-data and mixed-latency benchmark profiles.
pub fn run_enterprise_benchmarks() -> List(Benchmark) {
  let drivers = supported_drivers()

  [
    simulate_benchmark(postgres_async_driver(), HeavyData),
    simulate_benchmark(postgres_async_driver(), MixedLatency),
    simulate_benchmark(clickhouse_async_driver(), HeavyData),
    simulate_benchmark(clickhouse_async_driver(), MixedLatency),
    simulate_benchmark(kafka_async_driver(), HeavyData),
    simulate_benchmark(kafka_async_driver(), MixedLatency),
    simulate_benchmark(object_store_async_driver(), HeavyData),
    simulate_benchmark(object_store_async_driver(), MixedLatency),
  ]
  |> ensure_benchmark_driver_count(drivers)
}

/// Evaluate benchmarks against M48 budgets.
pub fn evaluate_budgets(
  benchmarks: List(Benchmark),
  budgets: List(Budget),
) -> List(BudgetResult) {
  evaluate_budgets_loop(benchmarks, budgets, [])
}

/// Count failing budget checks.
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 benchmark signature.
pub fn benchmark_signature(benchmark: Benchmark) -> String {
  benchmark.driver
  <> ":profile="
  <> profile_label(benchmark.profile)
  <> ":p50="
  <> int.to_string(benchmark.p50_latency_ms)
  <> ":p95="
  <> int.to_string(benchmark.p95_latency_ms)
  <> ":timeout_bps="
  <> int.to_string(benchmark.timeout_rate_bps)
  <> ":retry_bps="
  <> int.to_string(benchmark.retry_rate_bps)
  <> ":throughput="
  <> int.to_string(benchmark.throughput_ops_per_sec)
}

/// Stable budget-check result signature.
pub fn budget_result_signature(result: BudgetResult) -> String {
  result.driver
  <> ":profile="
  <> profile_label(result.profile)
  <> ":passed="
  <> bool_label(result.passed)
  <> ":reason="
  <> result.reason
}

/// Driver name accessor.
pub fn name(driver: DriverContract) -> String {
  driver.name
}

/// Driver telemetry accessor.
pub fn telemetry(driver: DriverContract) -> TelemetryProfile {
  driver.telemetry
}

/// Driver capabilities accessor.
pub fn capabilities(driver: DriverContract) -> List(DriverCapability) {
  driver.capabilities
}

fn timeout_policy(
  connect_ms: Int,
  request_ms: Int,
  idle_ms: Int,
) -> TimeoutPolicy {
  TimeoutPolicy(
    connect_ms: connect_ms,
    request_ms: request_ms,
    idle_ms: idle_ms,
  )
}

fn retry_policy(
  max_attempts: Int,
  base_backoff_ms: Int,
  max_backoff_ms: Int,
  jitter_percent: Int,
) -> RetryPolicy {
  RetryPolicy(
    max_attempts: max_attempts,
    base_backoff_ms: base_backoff_ms,
    max_backoff_ms: max_backoff_ms,
    jitter_percent: jitter_percent,
  )
}

fn telemetry_profile(driver_name: String) -> TelemetryProfile {
  TelemetryProfile(
    timeout_metric: "lightspeed.driver." <> driver_name <> ".timeouts_total",
    retry_metric: "lightspeed.driver." <> driver_name <> ".retries_total",
    latency_metric: "lightspeed.driver." <> driver_name <> ".latency_ms",
    in_flight_metric: "lightspeed.driver." <> driver_name <> ".in_flight",
  )
}

fn timeout_policy_valid(policy: TimeoutPolicy) -> Bool {
  policy.connect_ms > 0
  && policy.request_ms >= policy.connect_ms
  && policy.idle_ms >= policy.request_ms
}

fn retry_policy_valid(policy: RetryPolicy) -> Bool {
  policy.max_attempts >= 2
  && policy.base_backoff_ms > 0
  && policy.max_backoff_ms >= policy.base_backoff_ms
  && policy.jitter_percent >= 0
  && policy.jitter_percent <= 25
}

fn has_required_capabilities(capabilities: List(DriverCapability)) -> Bool {
  list.length(capabilities) >= 6
  && has_capability(capabilities, NonBlockingExecute)
  && has_capability(capabilities, BoundedConcurrency)
  && has_capability(capabilities, TimeoutTelemetry)
  && has_capability(capabilities, RetryTelemetry)
  && has_capability(capabilities, MixedLatencyTolerance)
}

fn has_capability(
  capabilities: List(DriverCapability),
  expected: DriverCapability,
) -> Bool {
  case capabilities {
    [] -> False
    [capability, ..rest] ->
      capability == expected || has_capability(rest, expected)
  }
}

fn telemetry_profile_valid(
  telemetry: TelemetryProfile,
  driver_name: String,
) -> Bool {
  telemetry.timeout_metric
  == "lightspeed.driver." <> driver_name <> ".timeouts_total"
  && telemetry.retry_metric
  == "lightspeed.driver." <> driver_name <> ".retries_total"
  && telemetry.latency_metric
  == "lightspeed.driver." <> driver_name <> ".latency_ms"
  && telemetry.in_flight_metric
  == "lightspeed.driver." <> driver_name <> ".in_flight"
}

fn all_drivers_valid(drivers: List(DriverContract)) -> Bool {
  case drivers {
    [] -> True
    [driver, ..rest] -> valid(driver) && all_drivers_valid(rest)
  }
}

fn timeout_retry_ranges_consistent(drivers: List(DriverContract)) -> Bool {
  case drivers {
    [] -> True
    [driver, ..rest] ->
      driver.timeout_policy.request_ms <= 800
      && driver.retry_policy.max_attempts <= 6
      && timeout_retry_ranges_consistent(rest)
  }
}

fn all_telemetry_metrics_present(drivers: List(DriverContract)) -> Bool {
  case drivers {
    [] -> True
    [driver, ..rest] -> {
      let telemetry = driver.telemetry
      telemetry.timeout_metric != ""
      && telemetry.retry_metric != ""
      && telemetry.latency_metric != ""
      && telemetry.in_flight_metric != ""
      && all_telemetry_metrics_present(rest)
    }
  }
}

fn telemetry_metric_suffixes_consistent(drivers: List(DriverContract)) -> Bool {
  case drivers {
    [] -> True
    [driver, ..rest] -> {
      let telemetry = driver.telemetry
      string.ends_with(telemetry.timeout_metric, ".timeouts_total")
      && string.ends_with(telemetry.retry_metric, ".retries_total")
      && string.ends_with(telemetry.latency_metric, ".latency_ms")
      && string.ends_with(telemetry.in_flight_metric, ".in_flight")
      && telemetry_metric_suffixes_consistent(rest)
    }
  }
}

fn timeout_policy_signature(policy: TimeoutPolicy) -> String {
  "connect="
  <> int.to_string(policy.connect_ms)
  <> ",request="
  <> int.to_string(policy.request_ms)
  <> ",idle="
  <> int.to_string(policy.idle_ms)
}

fn retry_policy_signature(policy: RetryPolicy) -> String {
  "attempts="
  <> int.to_string(policy.max_attempts)
  <> ",base="
  <> int.to_string(policy.base_backoff_ms)
  <> ",max="
  <> int.to_string(policy.max_backoff_ms)
  <> ",jitter="
  <> int.to_string(policy.jitter_percent)
}

fn telemetry_signature(profile: TelemetryProfile) -> String {
  "timeout="
  <> profile.timeout_metric
  <> ",retry="
  <> profile.retry_metric
  <> ",latency="
  <> profile.latency_metric
  <> ",in_flight="
  <> profile.in_flight_metric
}

fn simulate_benchmark(
  driver: DriverContract,
  profile: BenchmarkProfile,
) -> Benchmark {
  let #(base_p50, base_p95, base_timeout, base_retry, base_throughput) =
    class_baseline(driver.class)
  let #(p50_adjust, p95_adjust, timeout_adjust, retry_adjust, throughput_adjust) =
    profile_adjustment(profile)

  Benchmark(
    driver: driver.name,
    profile: profile,
    p50_latency_ms: base_p50 + p50_adjust,
    p95_latency_ms: base_p95 + p95_adjust,
    timeout_rate_bps: base_timeout + timeout_adjust,
    retry_rate_bps: base_retry + retry_adjust,
    throughput_ops_per_sec: base_throughput - throughput_adjust,
  )
}

fn class_baseline(class: DriverClass) -> #(Int, Int, Int, Int, Int) {
  case class {
    RelationalAdapter -> #(22, 58, 7, 18, 8600)
    ColumnarAdapter -> #(30, 76, 9, 22, 7800)
    StreamAdapter -> #(18, 50, 6, 16, 9800)
    ObjectStoreAdapter -> #(36, 92, 11, 26, 6500)
  }
}

fn profile_adjustment(profile: BenchmarkProfile) -> #(Int, Int, Int, Int, Int) {
  case profile {
    HeavyData -> #(12, 28, 4, 8, 1000)
    MixedLatency -> #(8, 20, 6, 10, 500)
  }
}

fn ensure_benchmark_driver_count(
  benchmarks: List(Benchmark),
  drivers: List(DriverContract),
) -> List(Benchmark) {
  case list.length(benchmarks) == list.length(drivers) * 2 {
    True -> benchmarks
    False -> []
  }
}

fn evaluate_budgets_loop(
  benchmarks: List(Benchmark),
  budgets: List(Budget),
  results_rev: List(BudgetResult),
) -> List(BudgetResult) {
  case benchmarks {
    [] -> list.reverse(results_rev)
    [benchmark, ..rest] -> {
      let result = evaluate_one_budget(benchmark, budgets)
      evaluate_budgets_loop(rest, budgets, [result, ..results_rev])
    }
  }
}

fn evaluate_one_budget(
  benchmark: Benchmark,
  budgets: List(Budget),
) -> BudgetResult {
  case budget_for_profile(benchmark.profile, budgets) {
    Error(reason) ->
      BudgetResult(
        driver: benchmark.driver,
        profile: benchmark.profile,
        passed: False,
        reason: reason,
      )
    Ok(budget) -> {
      let p95_ok = benchmark.p95_latency_ms <= budget.max_p95_latency_ms
      let timeout_ok = benchmark.timeout_rate_bps <= budget.max_timeout_rate_bps
      let retry_ok = benchmark.retry_rate_bps <= budget.max_retry_rate_bps
      let throughput_ok =
        benchmark.throughput_ops_per_sec >= budget.min_throughput_ops_per_sec
      let passed = p95_ok && timeout_ok && retry_ok && throughput_ok
      let reason = case passed {
        True -> "within_budget"
        False -> "budget_exceeded"
      }

      BudgetResult(
        driver: benchmark.driver,
        profile: benchmark.profile,
        passed: passed,
        reason: reason,
      )
    }
  }
}

fn budget_for_profile(
  profile: BenchmarkProfile,
  budgets: List(Budget),
) -> Result(Budget, String) {
  case budgets {
    [] -> Error("missing_budget_profile")
    [budget, ..rest] ->
      case budget.profile == profile {
        True -> Ok(budget)
        False -> budget_for_profile(profile, rest)
      }
  }
}

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)
  }
}