///|
pub(all) struct BulkheadConfig {
  max_concurrent : Int
  max_waiting : Int
} derive(Eq, Debug)

///|
pub(all) struct WaitingCall {
  call_id : String
  enqueued_at_ms : Int
} derive(Eq, Debug)

///|
pub(all) struct Bulkhead {
  config : BulkheadConfig
  active_calls : Array[String]
  waiting_calls : Array[WaitingCall]
  total_entered : Int
  total_queued : Int
  total_rejected : Int
  total_completed : Int
} derive(Eq, Debug)

///|
pub(all) enum BulkheadAdmission {
  BulkheadEntered(Bulkhead, Int)
  BulkheadQueued(Bulkhead, Int)
  BulkheadRejected(Bulkhead, String)
} derive(Eq, Debug)

///|
pub(all) struct BulkheadRelease {
  bulkhead : Bulkhead
  released : Bool
  promoted_call : String?
} derive(Eq, Debug)

///|
pub(all) struct BulkheadSnapshot {
  active : Int
  waiting : Int
  available : Int
  waiting_available : Int
  total_entered : Int
  total_queued : Int
  total_rejected : Int
  total_completed : Int
} derive(Eq, Debug)

///|
pub fn bulkhead_config(
  max_concurrent : Int,
  max_waiting : Int,
) -> BulkheadConfig {
  {
    max_concurrent: clamp_at_least_one(max_concurrent),
    max_waiting: clamp_non_negative(max_waiting),
  }
}

///|
pub fn new_bulkhead(config : BulkheadConfig) -> Bulkhead {
  {
    config: bulkhead_config(config.max_concurrent, config.max_waiting),
    active_calls: [],
    waiting_calls: [],
    total_entered: 0,
    total_queued: 0,
    total_rejected: 0,
    total_completed: 0,
  }
}

///|
pub fn default_bulkhead() -> Bulkhead {
  new_bulkhead(bulkhead_config(10, 20))
}

///|
pub fn bulkhead_admit(
  bulkhead : Bulkhead,
  call_id : String,
  now_ms : Int,
) -> BulkheadAdmission {
  if call_id.length() == 0 {
    return BulkheadRejected(
      { ..bulkhead, total_rejected: bulkhead.total_rejected + 1 },
      "call id must not be empty",
    )
  }
  if bulkhead_contains(bulkhead, call_id) {
    return BulkheadRejected(
      { ..bulkhead, total_rejected: bulkhead.total_rejected + 1 },
      "call id already exists",
    )
  }
  if bulkhead.active_calls.length() < bulkhead.config.max_concurrent {
    let active = bulkhead.active_calls.copy()
    active.push(call_id)
    BulkheadEntered(
      {
        ..bulkhead,
        active_calls: active,
        total_entered: bulkhead.total_entered + 1,
      },
      active.length(),
    )
  } else if bulkhead.waiting_calls.length() < bulkhead.config.max_waiting {
    let waiting = bulkhead.waiting_calls.copy()
    waiting.push({ call_id, enqueued_at_ms: clamp_non_negative(now_ms) })
    BulkheadQueued(
      {
        ..bulkhead,
        waiting_calls: waiting,
        total_queued: bulkhead.total_queued + 1,
      },
      waiting.length(),
    )
  } else {
    BulkheadRejected(
      { ..bulkhead, total_rejected: bulkhead.total_rejected + 1 },
      "concurrent and waiting capacity exhausted",
    )
  }
}

///|
pub fn bulkhead_complete(
  bulkhead : Bulkhead,
  call_id : String,
) -> BulkheadRelease {
  let active = remove_string(bulkhead.active_calls, call_id)
  if active.length() == bulkhead.active_calls.length() {
    return { bulkhead, released: false, promoted_call: None }
  }
  if bulkhead.waiting_calls.length() == 0 {
    return {
      bulkhead: {
        ..bulkhead,
        active_calls: active,
        total_completed: bulkhead.total_completed + 1,
      },
      released: true,
      promoted_call: None,
    }
  }
  let promoted = bulkhead.waiting_calls[0]
  let waiting = waiting_without_first(bulkhead.waiting_calls)
  active.push(promoted.call_id)
  {
    bulkhead: {
      ..bulkhead,
      active_calls: active,
      waiting_calls: waiting,
      total_entered: bulkhead.total_entered + 1,
      total_completed: bulkhead.total_completed + 1,
    },
    released: true,
    promoted_call: Some(promoted.call_id),
  }
}

///|
pub fn bulkhead_cancel_waiting(
  bulkhead : Bulkhead,
  call_id : String,
) -> Bulkhead {
  { ..bulkhead, waiting_calls: remove_waiting(bulkhead.waiting_calls, call_id) }
}

///|
pub fn bulkhead_wait_ms(
  bulkhead : Bulkhead,
  call_id : String,
  now_ms : Int,
) -> Int? {
  for waiting in bulkhead.waiting_calls {
    if waiting.call_id == call_id {
      return Some(max_int(0, now_ms - waiting.enqueued_at_ms))
    }
  }
  None
}

///|
pub fn bulkhead_contains(bulkhead : Bulkhead, call_id : String) -> Bool {
  if array_contains(bulkhead.active_calls, call_id) {
    return true
  }
  for waiting in bulkhead.waiting_calls {
    if waiting.call_id == call_id {
      return true
    }
  }
  false
}

///|
pub fn bulkhead_snapshot(bulkhead : Bulkhead) -> BulkheadSnapshot {
  {
    active: bulkhead.active_calls.length(),
    waiting: bulkhead.waiting_calls.length(),
    available: max_int(
      0,
      bulkhead.config.max_concurrent - bulkhead.active_calls.length(),
    ),
    waiting_available: max_int(
      0,
      bulkhead.config.max_waiting - bulkhead.waiting_calls.length(),
    ),
    total_entered: bulkhead.total_entered,
    total_queued: bulkhead.total_queued,
    total_rejected: bulkhead.total_rejected,
    total_completed: bulkhead.total_completed,
  }
}

///|
pub fn format_bulkhead_snapshot(snapshot : BulkheadSnapshot) -> String {
  "active=" +
  snapshot.active.to_string() +
  " waiting=" +
  snapshot.waiting.to_string() +
  " available=" +
  snapshot.available.to_string() +
  " rejected=" +
  snapshot.total_rejected.to_string() +
  " completed=" +
  snapshot.total_completed.to_string()
}

///|
fn remove_string(values : Array[String], target : String) -> Array[String] {
  let next : Array[String] = []
  for value in values {
    if value != target {
      next.push(value)
    }
  }
  next
}

///|
fn remove_waiting(
  values : Array[WaitingCall],
  target : String,
) -> Array[WaitingCall] {
  let next : Array[WaitingCall] = []
  for value in values {
    if value.call_id != target {
      next.push(value)
    }
  }
  next
}

///|
fn waiting_without_first(values : Array[WaitingCall]) -> Array[WaitingCall] {
  let next : Array[WaitingCall] = []
  for index = 1; index < values.length(); index = index + 1 {
    next.push(values[index])
  }
  next
}