persevero

//// `persevero` executes a fallible operation multiple times.
////
//// ```gleam
//// import gleam/http/request
//// import gleam/httpc
//// import gleam/io
//// import persevero
////
//// pub fn main() {
////   let assert Ok(request) = request.to("https://www.apple.com")
////
////   let response = {
////     use <- persevero.execute(
////       wait_stream: persevero.exponential_backoff(50, 2),
////       allow: persevero.all_errors,
////       mode: persevero.MaxAttempts(3),
////     )
////
////     httpc.send(request)
////   }
////
////   case response {
////     Ok(response) if response.status == 200 -> io.debug("Give me #prawducks. 😃")
////     _ -> io.debug("Guess I'll dev on Linux. 😔")
////   }
//// }
//// ```

import bigben/clock
import birl
import birl/duration
import gleam/erlang/process
import gleam/function
import gleam/int
import gleam/list
import gleam/yielder.{type Yielder}

/// Represents errors that can occur during execution attempts.
pub type Error(a) {
  /// Indicates that all execution attempts have been exhausted. Contains an
  /// ordered list of all errors encountered during the execution attempts.
  RetriesExhausted(errors: List(a))

  /// Indicates that the maximum duration for execution has been reached.
  /// Contains an ordered list of all errors encountered during the execution
  TimeExhausted(errors: List(a))

  /// Indicates that an error that wasn't allowed was encountered. Contains the
  /// specific error that caused execution to stop.
  UnallowedError(error: a)
}

type RetryResult(a, b) =
  Result(a, Error(b))

@internal
pub type RetryData(a, b) {
  RetryData(result: RetryResult(a, b), wait_times: List(Int), duration: Int)
}

/// Convenience function that you can supply to `execute`'s `allow` parameter to
/// allow all errors.
pub fn all_errors(_: a) -> Bool {
  True
}

/// Produces a custom wait stream.
pub fn custom_backoff(
  wait_time wait_time: Int,
  next_wait_time next_wait_time: fn(Int) -> Int,
) -> Yielder(Int) {
  yielder.iterate(wait_time, next_wait_time)
}

/// Produces a 0ms wait stream.
/// Ex: 0ms, 0ms, 0ms, ...
pub fn no_backoff() -> Yielder(Int) {
  yielder.repeat(0)
}

/// Produces a ms wait stream with a constant wait time.
/// Ex: 500ms, 500ms, 500ms, ...
pub fn constant_backoff(wait_time wait_time: Int) -> Yielder(Int) {
  yielder.iterate(wait_time, function.identity)
}

/// Produces a ms wait stream that increases linearly for each attempt.
/// Ex: 500ms, 1000ms, 1500ms, ...
pub fn linear_backoff(wait_time wait_time: Int, step step: Int) -> Yielder(Int) {
  yielder.iterate(wait_time, int.add(_, step))
}

/// Produces a ms wait stream that increases exponentially for each attempt.
/// time:
/// Ex: 500ms, 1000ms, 2000ms, 4000ms, ...
pub fn exponential_backoff(
  wait_time wait_time: Int,
  factor factor: Int,
) -> Yielder(Int) {
  yielder.iterate(wait_time, int.multiply(_, factor))
}

/// Adds a random integer between [1, `upper_bound`] to each wait time.
pub fn apply_jitter(
  wait_stream wait_stream: Yielder(Int),
  upper_bound upper_bound: Int,
) -> Yielder(Int) {
  apply_constant(wait_stream:, adjustment: int.random(upper_bound) + 1)
}

/// Adds a constant integer to each wait time.
pub fn apply_constant(
  wait_stream wait_stream: Yielder(Int),
  adjustment adjustment: Int,
) -> Yielder(Int) {
  wait_stream |> yielder.map(int.add(_, adjustment))
}

/// Multiplies each wait time by a constant factor.
pub fn apply_multiplier(
  wait_stream wait_stream: Yielder(Int),
  factor factor: Int,
) -> Yielder(Int) {
  wait_stream |> yielder.map(int.multiply(_, factor))
}

/// Caps each wait time at a maximum value.
pub fn apply_cap(
  wait_stream wait_stream: Yielder(Int),
  max_wait_time max_wait_time: Int,
) -> Yielder(Int) {
  wait_stream |> yielder.map(int.min(_, max_wait_time))
}

/// Configures the retry mode.
pub type Mode {
  /// Specifies the maximum number of attempts to make.
  MaxAttempts(Int)
  /// Specifies the maximum duration to make attempts for.
  Expiry(Int)
}

/// Initiates the execution process with the specified operation.
///
/// `allow` sets the logic for determining whether an error should trigger
/// another attempt. Expects a function that takes an error and returns a
/// boolean. Use this function to match on the encountered error and return
/// `True` for errors that should trigger another attempt, and `False` for
/// errors that should not. To allow all errors, use `all_errors`.
pub fn execute(
  wait_stream wait_stream: Yielder(Int),
  allow allow: fn(b) -> Bool,
  mode mode: Mode,
  operation operation: fn() -> Result(a, b),
) -> RetryResult(a, b) {
  execute_with_options(
    wait_stream:,
    allow:,
    mode:,
    operation: fn(_) { operation() },
    wait_function: process.sleep,
    clock: clock.new(),
  ).result
}

@internal
pub fn execute_with_options(
  wait_stream wait_stream: Yielder(Int),
  allow allow: fn(b) -> Bool,
  mode mode: Mode,
  operation operation: fn(Int) -> Result(a, b),
  wait_function wait_function: fn(Int) -> Nil,
  clock clock: clock.Clock,
) -> RetryData(a, b) {
  do_execute(
    wait_stream: wait_stream |> configure_wait_stream(mode),
    allow:,
    mode:,
    operation:,
    wait_function:,
    wait_time_acc: [],
    clock:,
    errors_acc: [],
    attempt_number: 0,
    start_time: clock |> clock.now(),
    duration: 0,
  )
}

fn configure_wait_stream(
  wait_stream wait_stream: Yielder(Int),
  mode mode: Mode,
) -> Yielder(Int) {
  let wait_stream =
    wait_stream
    |> yielder_prepend(0)
    |> yielder.map(int.max(_, 0))

  case mode {
    MaxAttempts(max_attempts) -> wait_stream |> yielder.take(max_attempts)
    Expiry(_) -> wait_stream
  }
}

// Swap to yielder.prepend when it's available.
// https://github.com/gleam-lang/yielder/pull/2
fn yielder_prepend(yielder: Yielder(a), element: a) -> Yielder(a) {
  use <- yielder.yield(element)
  yielder
}

fn do_execute(
  wait_stream wait_stream: Yielder(Int),
  allow allow: fn(b) -> Bool,
  mode mode: Mode,
  operation operation: fn(Int) -> Result(a, b),
  wait_function wait_function: fn(Int) -> Nil,
  wait_time_acc wait_time_acc: List(Int),
  clock clock: clock.Clock,
  errors_acc errors_acc: List(b),
  attempt_number attempt_number: Int,
  start_time start_time: birl.Time,
  duration duration: Int,
) -> RetryData(a, b) {
  let should_execute = case mode {
    MaxAttempts(max_attempts) -> max_attempts > 0
    Expiry(expiry) -> expiry > 0 && duration < expiry
  }

  case should_execute, wait_stream |> yielder.step() {
    True, yielder.Next(wait_time, wait_stream) -> {
      wait_function(wait_time)
      let wait_time_acc = [wait_time, ..wait_time_acc]

      let duration =
        clock
        |> clock.now()
        |> birl.difference(start_time)
        |> duration.blur_to(duration.MilliSecond)

      case operation(attempt_number) {
        Ok(result) ->
          RetryData(
            result: Ok(result),
            wait_times: wait_time_acc |> list.reverse,
            duration:,
          )
        Error(error) -> {
          case allow(error) {
            True ->
              do_execute(
                wait_stream:,
                allow:,
                mode:,
                operation:,
                wait_function:,
                wait_time_acc:,
                clock:,
                errors_acc: [error, ..errors_acc],
                attempt_number: attempt_number + 1,
                start_time: start_time,
                duration: duration,
              )
            False ->
              RetryData(
                result: Error(UnallowedError(error)),
                wait_times: wait_time_acc |> list.reverse,
                duration:,
              )
          }
        }
      }
    }
    _, _ -> {
      let error = case mode {
        MaxAttempts(_) -> RetriesExhausted(errors_acc |> list.reverse)
        Expiry(_) -> TimeExhausted(errors_acc |> list.reverse)
      }
      RetryData(
        result: Error(error),
        wait_times: wait_time_acc |> list.reverse,
        duration: duration,
      )
    }
  }
}