///|
/// Redis client.
///
/// Create one with `Client::Client`, run `Client::work` in a background async
/// task, then call command methods such as `get`, `set`, or `publish` from
/// other tasks.
///
/// Example:
/// ```moonbit nocheck
/// @async.with_task_group(group => {
///   let client = @redis.Client()
///   let worker = group.spawn(() => client.work(), allow_failure=true)
///   defer worker.cancel()
///
///   ignore(client.ping())
/// })
/// ```
struct Client {
  config : ClientConfig
  send_queue : @async.Queue[OutgoingCommand]
  receive_queue : @deque.Deque[PendingCommand]
  work_lock : @async.Semaphore
}

///|
/// Creates a client with optional connection settings.
///
/// Construction is cheap and does not connect to Redis. Start `Client::work`
/// before using regular command methods.
///
/// Example:
/// ```moonbit nocheck
/// let client = @redis.Client(
///   config=@redis.ClientConfig(host="127.0.0.1", port=6379),
/// )
/// ```
pub fn Client::Client(config? : ClientConfig = ClientConfig()) -> Client {
  {
    config,
    send_queue: @async.Queue::new(
      kind=match config.command_queue_max_length {
        None => Unbounded
        Some(length) => Blocking(length)
      },
    ),
    receive_queue: Deque([]),
    work_lock: @async.Semaphore::new(1),
  }
}

///|
async fn Client::dial(self : Client) -> @socket.Tcp {
  @async.with_timeout((self.config.connect_timeout * 1000).to_int(), () => {
    @socket.Tcp::connect_to_host(self.config.host, port=self.config.port)
  })
}

///|
async fn Client::handshake(
  self : Client,
  r : @resp.Reader,
  w : @resp.Writer,
) -> @resp.Version {
  // Default to RESP3, but fall back to RESP2 if the server doesn't support it.
  let hello : Array[Bytes] = [b"HELLO", b"3"]
  if self.config.password is Some(password) {
    hello.push(b"AUTH")
    hello.push(@encoding/utf8.encode(self.config.username))
    hello.push(@encoding/utf8.encode(password))
  }
  let version = match
    self.run_startup_command(
      w,
      r,
      ReadOnlyArray::from_array(hello),
      version=RESP3,
    ) {
    Ok(_) => @resp.RESP3
    Err(err) => {
      let msg = decode_utf8(err)
      guard is_hello_fallback_error(msg) else { raise HandshakeError(msg) }
      if self.config.password is Some(password) {
        self.expect_startup_ok(
          w,
          r,
          ReadOnlyArray::from_array(self.legacy_auth_command(password)),
          version=RESP2,
        )
      }
      RESP2
    }
  }
  if self.config.database != 0 {
    self.expect_startup_ok(
      w,
      r,
      [b"SELECT", @encoding/utf8.encode(self.config.database.to_string())],
      version~,
    )
  }
  if self.config.name is Some(name) {
    self.expect_startup_ok(
      w,
      r,
      [b"CLIENT", b"SETNAME", @encoding/utf8.encode(name)],
      version~,
    )
  }
  version
}

///|
fn Client::legacy_auth_command(
  self : Client,
  password : String,
) -> Array[Bytes] {
  let auth : Array[Bytes] = [b"AUTH"]
  if self.config.username != "default" {
    auth.push(@encoding/utf8.encode(self.config.username))
  }
  auth.push(@encoding/utf8.encode(password))
  auth
}

///|
async fn Client::run_startup_command(
  _self : Client,
  w : @resp.Writer,
  r : @resp.Reader,
  cmd : ReadOnlyArray[Bytes],
  version~ : @resp.Version,
) -> Result[@resp.Value, Bytes] {
  w.write_command(cmd)
  match r.read_response(version~) {
    Reply({ desc: Error(err), .. }) => Err(err)
    Reply(value) => Ok(value)
    Push(_) => raise @resp.ProtocolError
  }
}

///|
async fn Client::expect_startup_ok(
  self : Client,
  w : @resp.Writer,
  r : @resp.Reader,
  cmd : ReadOnlyArray[Bytes],
  version~ : @resp.Version,
) -> Unit {
  match self.run_startup_command(w, r, cmd, version~) {
    Ok(_) => ()
    Err(err) => raise HandshakeError(decode_utf8(err))
  }
}

///|
priv struct PendingCommand {
  args : ReadOnlyArray[Bytes]
  kind : PendingCommandKind
  mut state : PendingCommandState
  wakener : @promise.Wakener[Result[PendingCommandResponse, Error]]
}

///|
priv enum PendingCommandKind {
  Normal
}

///|
priv enum PendingCommandState {
  Queued
  Sent
  Cancelled
}

///|
priv enum PendingCommandResponse {
  Command(RawValue)
}

///|
priv enum OutgoingCommand {
  Pending(PendingCommand)
}

///|
fn PendingCommand::PendingCommand(
  args : ReadOnlyArray[Bytes],
  kind : PendingCommandKind,
  wakener : @promise.Wakener[Result[PendingCommandResponse, Error]],
) -> PendingCommand {
  { args, kind, state: Queued, wakener }
}

///|
fn PendingCommand::mark_sent(self : PendingCommand) -> Unit {
  if self.state is Queued {
    self.state = Sent
  }
}

///|
fn PendingCommand::cancel(self : PendingCommand) -> Unit {
  self.state = Cancelled
}

///|
fn PendingCommand::is_cancelled(self : PendingCommand) -> Bool {
  self.state is Cancelled
}

///|
/// Runs a custom Redis command through this client.
///
/// Use this when the package does not provide a typed helper for the command
/// you need. The command's decoder decides the result type and may raise if the
/// Redis response has an unexpected shape.
///
/// Example:
/// ```moonbit nocheck
/// let pong = client.execute(@redis.Command([b"PING"], value => value.as_string()))
/// ```
pub async fn[T] Client::execute(self : Client, cmd : Command[T]) -> T {
  let (waiter, wakener) = @promise.new()
  let pending = PendingCommand(
    ReadOnlyArray::from_array(cmd.args),
    Normal,
    wakener,
  )
  try {
    self.send_queue.put(Pending(pending))
    match waiter.wait() {
      Ok(Command(value)) => (cmd.res_decode_fn)(value)
      Err(err) => raise err
    }
  } catch {
    err => {
      if @async.is_cancellation_error(err) {
        pending.cancel()
      }
      raise err
    }
  }
}

///|
/// Keeps the regular command connection running.
///
/// A client may have at most one active work loop. Start a new `Client` if you
/// need another independent Redis connection. Cancelling the task running
/// `work` is the normal way to stop the client.
///
/// Example:
/// ```moonbit nocheck
/// @async.with_task_group(group => {
///   let client = @redis.Client()
///   let worker = group.spawn(() => client.work(), allow_failure=true)
///   defer worker.cancel()
///
///   assert_true(client.set("hello", "world"))
/// })
/// ```
pub async fn Client::work(self : Client) -> Unit {
  guard self.work_lock.try_acquire() else { raise ClientAlreadyWorking }
  defer self.work_lock.release()
  @async.with_task_group(group => {
    group.spawn_loop(
      () => self.run_connection(),
      retry=self.config.reconnect_strategy,
      fatal_error=err => err is HandshakeError(_),
    )
  }) catch {
    err => {
      self.send_queue.close(error=err, clear=false)
      self.flush_all_pending_commands(err)
      raise err
    }
  }
}

///|
fn is_hello_fallback_error(msg : String) -> Bool {
  let lower = msg.to_lower()
  (lower.contains("unknown command") && lower.contains("hello")) ||
  lower.contains("syntax error")
}

///|
async fn Client::run_connection(self : Client) -> Unit {
  let conn = self.dial()
  defer conn.close()
  @async.with_task_group(group => {
    let r = @resp.Reader(
      conn,
      buffer_size=self.config.read_buffer_size,
      max_depth=self.config.resp_max_depth,
    )
    let w = @resp.Writer(conn, buffer_size=self.config.write_buffer_size)
    let version = self.handshake(r, w)

    group.spawn_bg(() => {
      for ;; {
        let cmd = self.send_queue.get()
        match cmd {
          Pending(cmd) =>
            if !cmd.is_cancelled() {
              cmd.mark_sent()
              self.receive_queue.push_back(cmd)
              w.write_command(cmd.args)
            }
        }
      }
    })
    group.spawn_bg(() => {
      for ;; {
        match r.read_response(version~) {
          Reply({ desc: Error(err), .. }) => {
            let cmd = self.receive_queue.pop_front().unwrap()
            let err = response_error(err)
            if !cmd.is_cancelled() {
              cmd.wakener.wake(Err(err))
            }
          }
          Reply(val) => {
            let cmd = self.pop_normal_pending_command()
            if !cmd.is_cancelled() {
              cmd.wakener.wake(Ok(Command(val.raw())))
            }
          }
          Push(_values) => ()
        }
      }
    })
  }) catch {
    err => {
      self.flush_sent_pending_commands(err)
      raise err
    }
  }
}

///|
fn Client::pop_normal_pending_command(self : Client) -> PendingCommand {
  match self.receive_queue.front() {
    Some({ kind: Normal, .. }) => self.receive_queue.pop_front().unwrap()
    None => panic()
  }
}

///|
fn Client::flush_sent_pending_commands(self : Client, err : Error) -> Unit {
  let pending_err = pending_command_error(err)
  let cmds = self.receive_queue.drain(start=0)
  for cmd in cmds {
    if !cmd.is_cancelled() {
      cmd.wakener.wake(Err(pending_err))
    }
  }
}

///|
fn Client::flush_all_pending_commands(self : Client, err : Error) -> Unit {
  self.flush_sent_pending_commands(err)
  while (self.send_queue.try_get() catch { _ => None }) is Some(cmd) {
    match cmd {
      Pending(cmd) => if !cmd.is_cancelled() { cmd.wakener.wake(Err(err)) }
    }
  }
}

///|
fn pending_command_error(err : Error) -> Error {
  match err {
    ServerError(_)
    | TransportError(_)
    | HandshakeError(_)
    | UnexpectedResponse(_)
    | InvalidUtf8(_)
    | IntegerOverflow(_)
    | ClientAlreadyWorking => err
    _ => TransportError(err)
  }
}