// Connection settings shared by the service connection, lookups and redirects.

///|
/// Options for a Pulsar client. Durations are milliseconds.
pub struct ClientOptions {
  priv service_url : String
  priv auth : &Authentication?
  priv auth_provider : (async () -> AuthData)?
  priv connection_timeout_ms : Int
  priv operation_timeout_ms : Int
  priv keepalive_interval_ms : Int
  priv tls_insecure : Bool
  priv ca_pem_file : String?
  priv tls_identity : TlsIdentity?
  priv tls_identity_provider : (async () -> TlsIdentity)?
  priv tls_min_version : TlsVersion?
  priv tls_max_version : TlsVersion?
  priv tls_cipher_list : String?
  priv tls13_ciphersuites : String?
  priv listener_name : String?
  priv lookup_properties : Map[String, String]
  priv reconnect_policy : RetryPolicy
  priv max_connections_per_broker : Int
  priv connection_idle_timeout_ms : Int?
  priv max_memory_bytes : Int?
  priv on_event : ((ClientEvent) -> Unit)?
  priv on_span : ((ClientSpanEvent) -> Unit)?
}

///|
pub fn ClientOptions::new(
  service_url : String,
  auth? : &Authentication,
  auth_provider? : async () -> AuthData,
  connection_timeout_ms? : Int = 10000,
  operation_timeout_ms? : Int = 60000,
  keepalive_interval_ms? : Int = 30000,
  tls_insecure? : Bool = false,
  ca_pem_file? : String,
  tls_identity? : TlsIdentity,
  tls_identity_provider? : async () -> TlsIdentity,
  tls_min_version? : TlsVersion,
  tls_max_version? : TlsVersion,
  tls_cipher_list? : String,
  tls13_ciphersuites? : String,
  listener_name? : String,
  lookup_properties? : Map[String, String] = {},
  reconnect_policy? : RetryPolicy = RetryPolicy::default(),
  max_connections_per_broker? : Int = 1,
  connection_idle_timeout_ms? : Int,
  max_memory_bytes? : Int,
  on_event? : (ClientEvent) -> Unit,
  on_span? : (ClientSpanEvent) -> Unit,
) -> ClientOptions raise PulsarError {
  guard connection_timeout_ms > 0 &&
    operation_timeout_ms > 0 &&
    keepalive_interval_ms > 0 else {
    raise InvalidFrame("client timeouts must be positive")
  }
  guard !(auth is Some(_) && auth_provider is Some(_)) else {
    raise InvalidFrame("choose one authentication provider")
  }
  if ca_pem_file is Some(path) && path == "" {
    raise InvalidFrame("TLS CA path must not be empty")
  }
  guard !(tls_insecure && ca_pem_file is Some(_)) else {
    raise InvalidFrame("choose TLS verification or insecure TLS")
  }
  guard !(tls_identity is Some(_) && tls_identity_provider is Some(_)) else {
    raise InvalidFrame("choose TLS identity files or a refresh provider")
  }
  if tls_min_version is Some(minimum) &&
    tls_max_version is Some(maximum) &&
    minimum.native_value() > maximum.native_value() {
    raise InvalidFrame("TLS minimum version exceeds maximum version")
  }
  if tls_cipher_list == Some("") || tls13_ciphersuites == Some("") {
    raise InvalidFrame("TLS cipher policy must not be empty")
  }
  guard reconnect_policy.max_retries >= -1 &&
    reconnect_policy.initial_backoff_ms > 0 &&
    reconnect_policy.max_backoff_ms >= reconnect_policy.initial_backoff_ms else {
    raise InvalidFrame("invalid reconnect retry policy")
  }
  guard max_connections_per_broker > 0 else {
    raise InvalidFrame("max connections per broker must be positive")
  }
  if connection_idle_timeout_ms is Some(timeout) && timeout <= 0 {
    raise InvalidFrame("connection idle timeout must be positive")
  }
  if max_memory_bytes is Some(limit) && limit <= 0 {
    raise InvalidFrame("client memory limit must be positive")
  }
  {
    service_url,
    auth,
    auth_provider,
    connection_timeout_ms,
    operation_timeout_ms,
    keepalive_interval_ms,
    tls_insecure,
    ca_pem_file,
    tls_identity,
    tls_identity_provider,
    tls_min_version,
    tls_max_version,
    tls_cipher_list,
    tls13_ciphersuites,
    listener_name,
    lookup_properties,
    reconnect_policy,
    max_connections_per_broker,
    connection_idle_timeout_ms,
    max_memory_bytes,
    on_event,
    on_span,
  }
}

///|
/// Read a token each time a connection or authentication challenge needs it.
/// This permits token rotation without reconstructing the client.
pub fn token_file_provider(path : String) -> async () -> AuthData {
  async fn() {
    let token = @fs.read_file(path).text().trim().to_owned()
    guard token != "" else { raise InvalidFrame("token file is empty") }
    { method_name: "token", data: @utf8.encode(token), }
  }
}

///|
/// Use an Athenz role token supplied by a sidecar or another ZTS client.
/// The supplier is invoked on each connection and authentication challenge.
pub fn athenz_role_token_provider(
  supplier : async () -> String,
) -> async () -> AuthData {
  async fn() {
    let token = supplier().trim().to_owned()
    guard token != "" else {
      raise AuthenticationFailed("empty Athenz role token")
    }
    { method_name: "athenz", data: @utf8.encode(token), }
  }
}

///|
/// Read an Athenz role token file afresh for each connection or challenge.
pub fn athenz_role_token_file_provider(path : String) -> async () -> AuthData {
  athenz_role_token_provider(async fn() { @fs.read_file(path).text() })
}

///|
fn oauth2_form_component(value : String) -> String {
  let encoded = @buffer.Buffer()
  let digits = b"0123456789ABCDEF"
  for byte in @utf8.encode(value) {
    let code = byte.to_int()
    if (code >= 65 && code <= 90) ||
      (code >= 97 && code <= 122) ||
      (code >= 48 && code <= 57) ||
      code == 45 ||
      code == 46 ||
      code == 95 ||
      code == 126 {
      encoded.write_byte(byte)
    } else {
      encoded.write_byte(b'%')
      encoded.write_byte(digits[code >> 4])
      encoded.write_byte(digits[code & 15])
    }
  }
  @utf8.decode(encoded.to_bytes()) catch {
    _ => ""
  }
}

///|
/// OAuth2 client-credentials authentication using `client_secret_post`.
/// The provider requests a fresh access token on each connection or broker
/// challenge; pass it as `ClientOptions::new(auth_provider=...)`.
pub fn oauth2_client_credentials_provider(
  token_endpoint : String,
  client_id : String,
  client_secret : String,
  audience? : String,
  scope? : String,
) -> (async () -> AuthData) raise PulsarError {
  guard token_endpoint != "" && client_id != "" && client_secret != "" else {
    raise InvalidFrame("OAuth2 endpoint, client id and secret are required")
  }
  let mut form = "grant_type=client_credentials&client_id=" +
    oauth2_form_component(client_id) +
    "&client_secret=" +
    oauth2_form_component(client_secret)
  if audience is Some(value) {
    form += "&audience=" + oauth2_form_component(value)
  }
  if scope is Some(value) {
    form += "&scope=" + oauth2_form_component(value)
  }
  async fn() {
    let headers = Map([])
    headers[@http.CaseInsensitiveString("Content-Type")] = "application/x-www-form-urlencoded"
    let (response, data) = @http.request(
      token_endpoint,
      @http.RequestMethod::Post,
      headers,
      @utf8.encode(form),
    )
    guard response.code >= 200 && response.code < 300 else {
      raise ServerError(
        code=response.code,
        message="OAuth2 token endpoint rejected credentials",
      )
    }
    let payload = data.binary()
    let json = @json.parse(@utf8.decode(payload)) catch {
      _ => raise InvalidFrame("OAuth2 token response is not JSON")
    }
    guard json is Object(fields) &&
      fields.get("access_token") is Some(String(token)) &&
      token != "" else {
      raise InvalidFrame("OAuth2 token response has no access token")
    }
    { method_name: "token", data: @utf8.encode(token), }
  }
}