///|
/// The asynchronous half of the client.
///
/// Every public method has an `_async` twin here. The bodies are the same
/// request-building and decoding code as the synchronous versions; only the
/// hop through the transport changes. Keeping the two side by side rather
/// than sharing a generic core is deliberate: MoonBit cannot abstract over
/// "sync or async" without making the synchronous API pay for a future it
/// never uses, and the request construction is short enough that duplicating
/// the call sites is cheaper to read than a stack of higher-order helpers.
///
/// Use these with `AsyncTransport` (native and wasm); use the
/// synchronous methods with `Transport` (tests, and any host that
/// already drives its own I/O).

///|
/// Sends a request through an async transport.
///
/// Mirrors `ChromaClient::send`, including the pass-through of errors an
/// inner transport already classified.
async fn[T : AsyncTransport] ChromaClient::send_async(
  self : ChromaClient[T],
  verb : Method,
  path : String,
  body~ : Json = Json::null(),
) -> Response raise ChromaError {
  let headers = self.request_headers()
  let payload = match verb {
    Get | Delete => None
    _ => Some(body.stringify())
  }
  let request : Request = {
    verb,
    url: self.api_url(path),
    headers,
    body: payload,
  }
  let response = self.transport.send_async(request) catch {
    ChromaError::Api(_) as error => raise error
    ChromaError::Status(_, _) as error => raise error
    ChromaError::Decode(_) as error => raise error
    ChromaError::InvalidArgument(_) as error => raise error
    ChromaError::Transport(_) as error => raise error
    error => raise ChromaError::Transport(error.to_string())
  }
  if response.status < 200 || response.status >= 300 {
    raise api_error(response.status, response.body)
  }
  guard_response_body(response)
  response
}

///|
/// Sends a request and parses the response body as JSON.
async fn[T : AsyncTransport] ChromaClient::send_json_async(
  self : ChromaClient[T],
  verb : Method,
  path : String,
  body~ : Json = Json::null(),
) -> Json raise ChromaError {
  self.response_json_or_null(self.send_async(verb, path, body~))
}

///|
/// Checks that the server is reachable and returns its heartbeat timestamp.
pub async fn[T : AsyncTransport] ChromaClient::heartbeat_async(
  self : ChromaClient[T],
) -> Int64 raise ChromaError {
  self.heartbeat_value(self.send_json_async(Get, "heartbeat"))
}

///|
/// Returns the server version string.
pub async fn[T : AsyncTransport] ChromaClient::version_async(
  self : ChromaClient[T],
) -> String raise ChromaError {
  match self.send_json_async(Get, "version") {
    Json::String(s) => s
    _ => ""
  }
}

///|
/// Lists the collections in the configured tenant and database, with the same
/// optional paging as `list_collections`.
pub async fn[T : AsyncTransport] ChromaClient::list_collections_async(
  self : ChromaClient[T],
  limit~ : Int32 = 0,
  offset~ : Int32 = 0,
) -> Array[Collection] raise ChromaError {
  let response = self.send_json_async(
    Get,
    self.list_collections_path(limit, offset),
  )
  let items = match response {
    Json::Array(items) => items
    _ => raise ChromaError::Decode("expected an array of collections")
  }
  let out : Array[Collection] = []
  for item in items {
    out.push(Collection::from_json(item))
  }
  out
}

///|
/// Counts the collections in the configured tenant and database.
pub async fn[T : AsyncTransport] ChromaClient::count_collections_async(
  self : ChromaClient[T],
) -> Int32 raise ChromaError {
  let response = self.send_json_async(
    Get,
    self.database_prefix() + "/collections_count",
  )
  decode_count(response, "collections_count")
}

///|
/// Returns a collection handle for `name`, creating it when it is missing.
pub async fn[T : AsyncTransport] ChromaClient::get_or_create_collection_async(
  self : ChromaClient[T],
  name : String,
  config~ : CollectionConfig = CollectionConfig::new(),
) -> Collection raise ChromaError {
  let path = self.database_prefix() + "/collections"
  Collection::from_json(
    self.send_json_async(
      Post,
      path,
      body=create_body(name, config, get_or_create=true),
    ),
  )
}

///|
/// Creates a collection, failing when `name` is already taken.
pub async fn[T : AsyncTransport] ChromaClient::create_collection_async(
  self : ChromaClient[T],
  name : String,
  config~ : CollectionConfig = CollectionConfig::new(),
) -> Collection raise ChromaError {
  let path = self.database_prefix() + "/collections"
  Collection::from_json(
    self.send_json_async(
      Post,
      path,
      body=create_body(name, config, get_or_create=false),
    ),
  )
}

///|
/// Fetches an existing collection by name, returning `None` on 404.
pub async fn[T : AsyncTransport] ChromaClient::get_collection_async(
  self : ChromaClient[T],
  name : String,
) -> Collection? raise ChromaError {
  let path = self.database_prefix() +
    "/collections/" +
    encode_path_segment(name)
  let response = self.send_json_async(Get, path) catch {
    error =>
      if error.status_code() == Some(404) {
        return None
      } else {
        raise error
      }
  }
  Some(Collection::from_json(response))
}

///|
/// Deletes a collection by name.
pub async fn[T : AsyncTransport] ChromaClient::delete_collection_async(
  self : ChromaClient[T],
  name : String,
) -> Unit raise ChromaError {
  let path = self.database_prefix() +
    "/collections/" +
    encode_path_segment(name)
  ignore(self.send_json_async(Delete, path))
}

///|
/// Counts the records in a collection.
pub async fn[T : AsyncTransport] ChromaClient::count_async(
  self : ChromaClient[T],
  collection_id : String,
) -> Int32 raise ChromaError {
  let path = self.collection_path(collection_id) + "/count"
  decode_count(self.send_json_async(Get, path), "collections/{id}/count")
}

///|
/// Adds records to a collection.
pub async fn[T : AsyncTransport] ChromaClient::add_async(
  self : ChromaClient[T],
  collection_id : String,
  records : Array[Record],
  options~ : WriteOptions = WriteOptions::new(),
  on_chunk~ : ChunkProgress = no_progress,
) -> Unit raise ChromaError {
  self.write_chunks_async(
    collection_id,
    records,
    "add",
    options,
    reject_duplicate_ids=true,
    on_chunk~,
  )
}

///|
/// The async half of `write_chunks`: same validation, same chunk sizes, one
/// request per chunk, each awaited before the next is built.
///
/// It does not go through `write_records` because that helper takes a
/// synchronous sender; the chunk loop is spelled out here so a failure in
/// chunk `n` still stops chunk `n+1`, which is the property a retry depends
/// on.
async fn[T : AsyncTransport] ChromaClient::write_chunks_async(
  self : ChromaClient[T],
  collection_id : String,
  records : Array[Record],
  endpoint : String,
  options : WriteOptions,
  reject_duplicate_ids~ : Bool,
  on_chunk~ : ChunkProgress = no_progress,
) -> Unit raise ChromaError {
  let path = self.collection_path(collection_id) + "/" + endpoint
  let sent : Ref[Int32] = Ref(0)
  let total = records.length()
  let sender : AsyncChunkSender = chunk => {
    ignore(self.send_json_async(Post, path, body=records_to_columns(chunk)))
    sent.val = sent.val + chunk.length()
    on_chunk(sent.val, total)
  }
  write_records_async(records, options, reject_duplicate_ids, sender)
}

///|
/// Inserts or replaces records by id.
pub async fn[T : AsyncTransport] ChromaClient::upsert_async(
  self : ChromaClient[T],
  collection_id : String,
  records : Array[Record],
  options~ : WriteOptions = WriteOptions::new(),
  on_chunk~ : ChunkProgress = no_progress,
) -> Unit raise ChromaError {
  self.write_chunks_async(
    collection_id,
    records,
    "upsert",
    options,
    reject_duplicate_ids=false,
    on_chunk~,
  )
}

///|
/// Updates the document, metadata or embedding of existing records.
///
/// Records are matched by id; ids that do not exist are ignored by Chroma.
pub async fn[T : AsyncTransport] ChromaClient::update_async(
  self : ChromaClient[T],
  collection_id : String,
  records : Array[Record],
  filter~ : Filter = Filter::new(),
) -> Unit raise ChromaError {
  // An update patches the records it names, so a missing embedding or
  // document is the normal case rather than an error: a `null` column entry
  // tells Chroma to leave that field alone.
  validate_records(records, require_embedding=false)
  validate_batch(records, reject_duplicate_ids=false)
  // Same body as `update`: records plus the clauses that pick them, with no
  // `include` / `offset` — a write returns nothing and pages nothing.
  let body : Map[String, Json] = match records_to_columns(records) {
    Json::Object(fields) => fields.copy()
    _ => Map([])
  }
  for entry in update_clauses(filter) {
    let (key, value) = entry
    body[key] = value
  }
  let path = self.collection_path(collection_id) + "/update"
  ignore(self.send_json_async(Post, path, body=Json::object(body)))
}

///|
/// Deletes the records a filter names. Same rules as `delete`: at least one
/// of `ids`, `where` or `where_document` is required.
pub async fn[T : AsyncTransport] ChromaClient::delete_async(
  self : ChromaClient[T],
  collection_id : String,
  filter~ : Filter = Filter::new(),
) -> Unit raise ChromaError {
  let body = delete_body(filter)
  let path = self.collection_path(collection_id) + "/delete"
  ignore(self.send_json_async(Post, path, body~))
}

///|
/// Fetches records by filter, or the whole collection when the filter is
/// empty.
///
/// The async counterpart of `get`: `include_fields` defaults to
/// `Include::get_default()` (documents and metadata), not to the query default,
/// because Chroma rejects a `POST /get` that carries `distances`.
pub async fn[T : AsyncTransport] ChromaClient::get_async(
  self : ChromaClient[T],
  collection_id : String,
  filter~ : Filter = Filter::new(),
  include_fields~ : Include = Include::get_default(),
) -> QueryResult raise ChromaError {
  validate_limit_offset(filter.limit, filter.offset)
  let path = self.collection_path(collection_id) + "/get"
  QueryResult::from_json(
    self.send_json_async(Post, path, body=filter.to_json(include_fields)),
  )
}

///|
/// Searches a collection with caller-supplied query vectors.
pub async fn[T : AsyncTransport] ChromaClient::query_embeddings_async(
  self : ChromaClient[T],
  collection_id : String,
  embeddings : Array[Embedding],
  n_results~ : Int32 = 10,
  include_fields~ : Include = Include::default(),
  filter~ : Filter = Filter::new(),
) -> QueryResult raise ChromaError {
  let body = query_body(
    embeddings,
    n_results,
    include_fields,
    filter,
    filter.limit,
    filter.offset,
  )
  let path = self.collection_path(collection_id) + "/query"
  let response = self.send_async(Post, path, body~)
  QueryResult::from_query_json(self.response_json_or_null(response))
}

///|
/// Returns a small sample of the records in a collection.
///
/// The async counterpart of `peek`: same body, same clamp, same defaults —
/// `limit` is clamped to `PEEK_LIMIT` (1..=100) and `include_fields` defaults
/// to `Include::get_default()` (documents and metadata). See `peek` for why
/// there is no `where` clause.
pub async fn[T : AsyncTransport] ChromaClient::peek_async(
  self : ChromaClient[T],
  collection_id : String,
  limit~ : Int32 = 10,
  include_fields~ : Include = Include::get_default(),
) -> QueryResult raise ChromaError {
  let body = peek_body(limit, include_fields)
  let path = self.collection_path(collection_id) + "/get"
  QueryResult::from_json(self.send_json_async(Post, path, body~))
}

///|
/// Updates a collection's name and/or metadata.
///
/// The async counterpart of `modify_collection`: `new_name` of `None` keeps
/// the current name, `metadata` of `None` keeps the current metadata, and
/// `name` may be the collection's id or its name, which is looked up first
/// because the server's `PUT` endpoint only accepts an id.
pub async fn[T : AsyncTransport] ChromaClient::modify_collection_async(
  self : ChromaClient[T],
  name : String,
  new_name~ : String? = None,
  metadata~ : Metadata? = None,
) -> Unit raise ChromaError {
  let body = modify_body(new_name, metadata)
  let id = self.resolve_collection_id_async(name)
  let path = self.database_prefix() + "/collections/" + encode_path_segment(id)
  ignore(self.send_json_async(Put, path, body=Json::object(body)))
}

///|
/// The async counterpart of `create_database`.
pub async fn[T : AsyncTransport] ChromaClient::create_database_async(
  self : ChromaClient[T],
) -> Unit raise ChromaError {
  ignore(
    self.send_json_async(
      Post,
      self.create_database_path(),
      body=self.create_database_body(),
    ),
  )
}

///|
/// The async counterpart of `resolve_collection_id`.
async fn[T : AsyncTransport] ChromaClient::resolve_collection_id_async(
  self : ChromaClient[T],
  name_or_id : String,
) -> String raise ChromaError {
  if is_uuid_v4(name_or_id) {
    return name_or_id
  }
  match self.get_collection_async(name_or_id) {
    Some(collection) => collection.id
    None =>
      raise ChromaError::InvalidArgument("no such collection: " + name_or_id)
  }
}