// The search commands. An index names a set of documents — hash keys or JSON documents
// under a prefix — and a schema says which fields are searchable, sortable, or a vector.
// `FT.SEARCH` answers with documents, `FT.AGGREGATE` with computed rows, and `FT.HYBRID`
// fuses a text search and a vector search into one ranking.
// https://redis.io/docs/latest/commands/?group=search

///|
/// One value in a search or aggregate reply: the fields of a document as text, and the
/// computed values of an aggregation as the type the server sent.
#derive(Eq, Debug)
pub(all) enum FieldValue {
  Text(String)
  Integer(Int64)
  Number(Double)
  Null
}

///|
pub extend FieldValue with Debug::{to_repr}

///|
pub extend FieldValue with Eq::{not_equal, equal}

///|
/// A reply value as a `FieldValue`.
pub fn field_value(reply : @resp.RespValue) -> FieldValue {
  match reply {
    @resp.Integer(n) => Integer(n)
    @resp.Double(d) => Number(d)
    @resp.SimpleString(s) => Text(s)
    @resp.BulkString(b) | @resp.VerbatimString(_, b) =>
      Text(@utf8.decode_lossy(b[:]))
    _ => Null
  }
}

///|
/// A value as text, for the callers that only ever see text.
pub fn FieldValue::text(self : FieldValue) -> String? {
  match self {
    Text(s) => Some(s)
    Integer(n) => Some(n.to_string())
    Number(d) => Some(d.to_string())
    Null => None
  }
}

///|
/// A value as an integer, when it holds one.
pub fn FieldValue::int(self : FieldValue) -> Int64? {
  match self {
    Integer(n) => Some(n)
    Number(d) => Some(d.to_int64())
    Text(s) => {
      let n = @string.parse_int64(s) catch { _ => return None }
      Some(n)
    }
    Null => None
  }
}

///|
/// A value as a double, when it holds one.
pub fn FieldValue::double(self : FieldValue) -> Double? {
  match self {
    Integer(n) => Some(n.to_double())
    Number(d) => Some(d)
    Text(s) => {
      let d = @string.parse_double(s) catch { _ => return None }
      Some(d)
    }
    Null => None
  }
}

///|
/// One document a query returned: its key, the fields asked for, and the score, payload or
/// sort key when the query asked for them.
pub(all) struct SearchHit {
  id : String
  fields : Array[(String, @resp.RespValue)]
  score : Double?
  payload : Bytes?
  sort_key : String?
}

///|
/// The field `name` of a hit as text, or `None` when the query did not return it.
pub fn SearchHit::field(self : SearchHit, name : String) -> String? {
  for (field, value) in self.fields {
    if field == name {
      return field_value(value).text()
    }
  }
  None
}

///|
/// What `FT.SEARCH` and `FT.HYBRID` answer: how many documents matched, the ones this page
/// returned, and any warning the query raised.
pub(all) struct SearchResult {
  /// The total number of matches, before `LIMIT`.
  total : Int64
  hits : Array[SearchHit]
  warnings : Array[String]
}

///|
/// What `FT.AGGREGATE` and `FT.CURSOR READ` answer: the rows, and the cursor id to read
/// more with.
pub(all) struct AggregateResult {
  total : Int64
  /// Each row is the field/value pairs in the order the server produced them.
  rows : Array[Array[(String, FieldValue)]]
  /// The id to read more with `ft_cursor_read`, or `None` when the reply came whole.
  cursor : Int64?
}

///|
/// The value of `name` in an aggregate row.
pub fn aggregate_field(
  row : Array[(String, FieldValue)],
  name : String,
) -> FieldValue? {
  for (field, value) in row {
    if field == name {
      return Some(value)
    }
  }
  None
}

///|
/// A field to sort by and the direction.
pub(all) struct SortKey {
  field : String
  /// True for `ASC`, false for `DESC`.
  ascending : Bool
}

///|
/// Which fragments of each field `SUMMARIZE` returns, and with what separator.
pub(all) struct Summarize {
  fields : Array[String]
  /// The number of fragments per field; the server picks 3 by default.
  fragments : Int64?
  /// The length of each fragment in words.
  length : Int64?
  separator : String?
}

///|
/// A numeric range a document's field must fall in.
pub(all) struct NumericFilter {
  field : String
  /// The lower bound as text, so `-inf`, `+inf` and the exclusive `(1` can be written.
  min : String
  /// The upper bound as text.
  max : String
}

///|
/// A distance a document's geo field must fall within.
pub(all) struct GeoFilter {
  field : String
  longitude : Double
  latitude : Double
  radius : Double
  /// `m`, `km`, `mi` or `ft`.
  unit : String
}

///|
/// The options `FT.SEARCH` takes besides the index and the query.
pub(all) struct SearchOptions {
  /// Return only the ids.
  nocontent : Bool
  /// Do not stem the query terms.
  verbatim : Bool
  with_scores : Bool
  with_payloads : Bool
  with_sort_keys : Bool
  /// The fields to return; empty returns them all.
  return_fields : Array[String]
  /// The fields to wrap matches in `` tags in; empty leaves them alone.
  highlight : Array[String]
  summarize : Summarize?
  sort_by : SortKey?
  /// How many matches to skip; sent only with `limit`.
  offset : Int64
  /// How many matches to return; 0 leaves the server's default of 10.
  limit : Int64
  /// The query dialect, 1 to 4; `None` leaves the server default.
  dialect : Int32?
  /// `PARAMS` values, by the name the query refers to. A vector is raw bytes.
  params : Array[(String, Bytes)]
  timeout : Int64?
  /// Restrict the search to these keys.
  in_keys : Array[String]
  /// Restrict the search to these fields.
  in_fields : Array[String]
  filter : NumericFilter?
  geo_filter : GeoFilter?
  /// `TFIDF`, `BM25STD`, `DISMAX` or a scorer a module registered.
  scorer : String?
  /// How far apart the query terms may be, in words.
  slop : Int64?
  language : String?
  /// Return the score breakdown alongside each hit.
  explain_score : Bool
}

///|
/// `FT.SEARCH`'s options at their documented defaults.
pub let search_default : SearchOptions = SearchOptions::{
  nocontent: false,
  verbatim: false,
  with_scores: false,
  with_payloads: false,
  with_sort_keys: false,
  return_fields: [],
  highlight: [],
  summarize: None,
  sort_by: None,
  offset: 0,
  limit: 0,
  dialect: None,
  params: [],
  timeout: None,
  in_keys: [],
  in_fields: [],
  filter: None,
  geo_filter: None,
  scorer: None,
  slop: None,
  language: None,
  explain_score: false,
}

///|
/// One `REDUCE` of an aggregation: a function, its arguments, and the name its result
/// takes.
pub(all) struct Reducer {
  /// `COUNT`, `COUNT_DISTINCT`, `SUM`, `AVG`, `MIN`, `MAX`, `QUANTILE`, `STDDEV`,
  /// `TOLIST`, `FIRST_VALUE` or `RANDOM_SAMPLE`.
  function : String
  /// The arguments as the query language writes them: `0` for `COUNT`, `@price` for `SUM`,
  /// `2 @price 0.5` for `QUANTILE`.
  args : Array[String]
  as_ : String?
}

///|
/// The options `FT.AGGREGATE` takes besides the index and the query.
pub(all) struct AggregateOptions {
  verbatim : Bool
  /// The fields to carry into the result; `["*"]` loads all of them.
  load : Array[String]
  group_by : Array[String]
  reducers : Array[Reducer]
  /// `APPLY` steps: an expression and the name its result takes.
  apply : Array[(String, String)]
  sort_by : Array[SortKey]
  offset : Int64
  /// 0 leaves the server's default of 10.
  limit : Int64
  /// A post-aggregation filter expression.
  filter : String?
  params : Array[(String, Bytes)]
  dialect : Int32?
  timeout : Int64?
  /// Ask for a cursor holding this many rows per read.
  cursor_count : Int64?
  /// Close an idle cursor after this many milliseconds.
  max_idle : Int64?
}

///|
/// `FT.AGGREGATE`'s options at their documented defaults.
pub let aggregate_default : AggregateOptions = AggregateOptions::{
  verbatim: false,
  load: [],
  group_by: [],
  reducers: [],
  apply: [],
  sort_by: [],
  offset: 0,
  limit: 0,
  filter: None,
  params: [],
  dialect: None,
  timeout: None,
  cursor_count: None,
  max_idle: None,
}

///|
/// What an index is built over.
#derive(Eq, Debug)
pub(all) enum IndexKind {
  Hash
  Json
}

///|
pub extend IndexKind with Debug::{to_repr}

///|
pub extend IndexKind with Eq::{not_equal, equal}

///|
/// The type of one schema field.
#derive(Eq, Debug)
pub(all) enum SchemaKind {
  Text
  Tag
  Numeric
  Geo
  Vector
}

///|
pub extend SchemaKind with Debug::{to_repr}

///|
pub extend SchemaKind with Eq::{not_equal, equal}

///|
/// How a vector field is indexed: `Flat` scans every vector, `Hnsw` builds a navigable
/// small-world graph over them.
#derive(Eq, Debug)
pub(all) enum VectorAlgorithm {
  Flat
  Hnsw
}

///|
pub extend VectorAlgorithm with Debug::{to_repr}

///|
pub extend VectorAlgorithm with Eq::{not_equal, equal}

///|
/// The parameters a `VECTOR` field takes.
pub(all) struct VectorSpec {
  algorithm : VectorAlgorithm
  /// How many dimensions each vector has.
  dim : Int32
  /// `FLOAT32`, `FLOAT64`, `BFLOAT16`, `FLOAT16`, `INT8` or `UINT8`.
  element : String
  /// `L2`, `IP` or `COSINE`.
  distance : String
  initial_cap : Int64?
  block_size : Int64?
  /// `HNSW` only: edges per node.
  m : Int64?
  /// `HNSW` only: the candidate list size during construction.
  ef_construction : Int64?
  /// `HNSW` only: the candidate list size during a query.
  ef_runtime : Int64?
  /// `HNSW` only: the range query precision.
  epsilon : Double?
}

///|
/// A flat (brute force) vector index over `dim` dimensions.
pub fn VectorSpec::flat(
  dim : Int32,
  element? : String,
  distance? : String,
  initial_cap? : Int64,
  block_size? : Int64,
) -> VectorSpec {
  {
    algorithm: Flat,
    dim,
    element: element.unwrap_or("FLOAT32"),
    distance: distance.unwrap_or("L2"),
    initial_cap,
    block_size,
    m: None,
    ef_construction: None,
    ef_runtime: None,
    epsilon: None,
  }
}

///|
/// An HNSW vector index over `dim` dimensions.
pub fn VectorSpec::hnsw(
  dim : Int32,
  element? : String,
  distance? : String,
  initial_cap? : Int64,
  m? : Int64,
  ef_construction? : Int64,
  ef_runtime? : Int64,
  epsilon? : Double,
) -> VectorSpec {
  {
    algorithm: Hnsw,
    dim,
    element: element.unwrap_or("FLOAT32"),
    distance: distance.unwrap_or("L2"),
    initial_cap,
    block_size: None,
    m,
    ef_construction,
    ef_runtime,
    epsilon,
  }
}

///|
/// One field of a schema: the document field it reads, the name queries use, and the
/// options its type takes.
pub(all) struct SchemaField {
  /// The hash field, or the JSONPath for a JSON index.
  name : String
  kind : SchemaKind
  /// The name the field takes in queries, when it differs from `name`.
  as_ : String?
  sortable : Bool
  /// Sort by the raw value rather than a normalized one.
  unf : Bool
  noindex : Bool
  /// `TEXT` only: the term weight.
  weight : Double?
  /// `TAG` only: the separator between tags, `,` by default.
  separator : String?
  /// `TAG` only: match tags case-sensitively.
  case_sensitive : Bool
  /// `TEXT` only: the phonetic matcher, `dm:en` or `dm:fr`.
  phonetic : String?
  /// `TEXT` only: index the term without stemming it.
  no_stem : Bool
  with_suffix_trie : Bool
  index_empty : Bool
  index_missing : Bool
  /// `VECTOR` only: the algorithm and its parameters.
  vector : VectorSpec?
}

///|
/// A field of the given kind with every option off.
fn schema_field(name : String, kind : SchemaKind, as_ : String?) -> SchemaField {
  {
    name,
    kind,
    as_,
    sortable: false,
    unf: false,
    noindex: false,
    weight: None,
    separator: None,
    case_sensitive: false,
    phonetic: None,
    no_stem: false,
    with_suffix_trie: false,
    index_empty: false,
    index_missing: false,
    vector: None,
  }
}

///|
/// A `TEXT` field.
pub fn SchemaField::text(
  name : String,
  as_? : String,
  sortable? : Bool,
  unf? : Bool,
  noindex? : Bool,
  weight? : Double,
  phonetic? : String,
  no_stem? : Bool,
  with_suffix_trie? : Bool,
  index_empty? : Bool,
  index_missing? : Bool,
) -> SchemaField {
  {
    ..schema_field(name, Text, as_),
    sortable: sortable.unwrap_or(false),
    unf: unf.unwrap_or(false),
    noindex: noindex.unwrap_or(false),
    weight,
    phonetic,
    no_stem: no_stem.unwrap_or(false),
    with_suffix_trie: with_suffix_trie.unwrap_or(false),
    index_empty: index_empty.unwrap_or(false),
    index_missing: index_missing.unwrap_or(false),
  }
}

///|
/// A `TAG` field.
pub fn SchemaField::tag(
  name : String,
  as_? : String,
  separator? : String,
  case_sensitive? : Bool,
  sortable? : Bool,
  unf? : Bool,
  noindex? : Bool,
  with_suffix_trie? : Bool,
  index_empty? : Bool,
  index_missing? : Bool,
) -> SchemaField {
  {
    ..schema_field(name, Tag, as_),
    separator,
    case_sensitive: case_sensitive.unwrap_or(false),
    sortable: sortable.unwrap_or(false),
    unf: unf.unwrap_or(false),
    noindex: noindex.unwrap_or(false),
    with_suffix_trie: with_suffix_trie.unwrap_or(false),
    index_empty: index_empty.unwrap_or(false),
    index_missing: index_missing.unwrap_or(false),
  }
}

///|
/// A `NUMERIC` field.
pub fn SchemaField::numeric(
  name : String,
  as_? : String,
  sortable? : Bool,
  unf? : Bool,
  noindex? : Bool,
  index_empty? : Bool,
  index_missing? : Bool,
) -> SchemaField {
  {
    ..schema_field(name, Numeric, as_),
    sortable: sortable.unwrap_or(false),
    unf: unf.unwrap_or(false),
    noindex: noindex.unwrap_or(false),
    index_empty: index_empty.unwrap_or(false),
    index_missing: index_missing.unwrap_or(false),
  }
}

///|
/// A `GEO` field.
pub fn SchemaField::geo(
  name : String,
  as_? : String,
  sortable? : Bool,
  unf? : Bool,
  noindex? : Bool,
  index_empty? : Bool,
  index_missing? : Bool,
) -> SchemaField {
  {
    ..schema_field(name, Geo, as_),
    sortable: sortable.unwrap_or(false),
    unf: unf.unwrap_or(false),
    noindex: noindex.unwrap_or(false),
    index_empty: index_empty.unwrap_or(false),
    index_missing: index_missing.unwrap_or(false),
  }
}

///|
/// A `VECTOR` field.
pub fn SchemaField::vector(
  name : String,
  spec : VectorSpec,
  as_? : String,
  sortable? : Bool,
  noindex? : Bool,
) -> SchemaField {
  {
    ..schema_field(name, Vector, as_),
    sortable: sortable.unwrap_or(false),
    noindex: noindex.unwrap_or(false),
    vector: Some(spec),
  }
}

///|
/// What `FT.CREATE` builds: the documents it reads and the fields it indexes.
pub(all) struct Schema {
  /// Hash keys or JSON documents.
  on : IndexKind
  /// The key prefixes the index reads; empty indexes every key.
  prefixes : Array[String]
  /// A filter expression a document must match to be indexed.
  filter : String?
  language : String?
  language_field : String?
  score : Double?
  score_field : String?
  /// Build a temporary index that expires after this many seconds.
  temporary : Int64?
  /// The stop words; `None` leaves the server's list, an empty array disables them.
  stopwords : Array[String]?
  fields : Array[SchemaField]
}

///|
/// A schema over `fields` with the default index options.
pub fn Schema::new(fields : Array[SchemaField]) -> Schema {
  {
    on: Hash,
    prefixes: [],
    filter: None,
    language: None,
    language_field: None,
    score: None,
    score_field: None,
    temporary: None,
    stopwords: None,
    fields,
  }
}

///|
/// Append the keywords that describe one field.
fn encode_schema_field(args : Array[Bytes], field : SchemaField) -> Unit {
  args.push(arg_string(field.name))
  push_some(args, b"AS", field.as_.map(arg_string))
  match field.kind {
    Text => {
      args.push(b"TEXT")
      push_some(args, b"WEIGHT", field.weight.map(arg_double))
      if field.no_stem {
        args.push(b"NOSTEM")
      }
      push_some(args, b"PHONETIC", field.phonetic.map(arg_string))
    }
    Tag => {
      args.push(b"TAG")
      push_some(args, b"SEPARATOR", field.separator.map(arg_string))
      if field.case_sensitive {
        args.push(b"CASESENSITIVE")
      }
    }
    Numeric => args.push(b"NUMERIC")
    Geo => args.push(b"GEO")
    Vector => {
      args.push(b"VECTOR")
      match field.vector {
        Some(spec) => encode_vector_spec(args, spec)
        None => ()
      }
    }
  }
  if field.with_suffix_trie {
    args.push(b"WITHSUFFIXTRIE")
  }
  if field.index_empty {
    args.push(b"INDEXEMPTY")
  }
  if field.index_missing {
    args.push(b"INDEXMISSING")
  }
  if field.sortable {
    args.push(b"SORTABLE")
    if field.unf {
      args.push(b"UNF")
    }
  }
  if field.noindex {
    args.push(b"NOINDEX")
  }
}

///|
/// Append `FLAT|HNSW  TYPE ... DIM ... DISTANCE_METRIC ...`, the count being the
/// number of tokens that follow the algorithm.
fn encode_vector_spec(args : Array[Bytes], spec : VectorSpec) -> Unit {
  let pairs : Array[Bytes] = [
    b"TYPE",
    arg_string(spec.element),
    b"DIM",
    arg_int(spec.dim),
    b"DISTANCE_METRIC",
    arg_string(spec.distance),
  ]
  match spec.initial_cap {
    Some(n) => {
      pairs.push(b"INITIAL_CAP")
      pairs.push(arg_int64(n))
    }
    None => ()
  }
  match spec.algorithm {
    Flat => {
      match spec.block_size {
        Some(n) => {
          pairs.push(b"BLOCK_SIZE")
          pairs.push(arg_int64(n))
        }
        None => ()
      }
      args.push(b"FLAT")
      args.push(arg_int(pairs.length()))
      for pair in pairs {
        args.push(pair)
      }
    }
    Hnsw => {
      match spec.m {
        Some(n) => {
          pairs.push(b"M")
          pairs.push(arg_int64(n))
        }
        None => ()
      }
      match spec.ef_construction {
        Some(n) => {
          pairs.push(b"EF_CONSTRUCTION")
          pairs.push(arg_int64(n))
        }
        None => ()
      }
      match spec.ef_runtime {
        Some(n) => {
          pairs.push(b"EF_RUNTIME")
          pairs.push(arg_int64(n))
        }
        None => ()
      }
      match spec.epsilon {
        Some(d) => {
          pairs.push(b"EPSILON")
          pairs.push(arg_double(d))
        }
        None => ()
      }
      args.push(b"HNSW")
      args.push(arg_int(pairs.length()))
      for pair in pairs {
        args.push(pair)
      }
    }
  }
}

///|
/// `FT.CREATE index [ON HASH|JSON] [PREFIX n ...] [FILTER expr] [LANGUAGE l]
/// [LANGUAGE_FIELD f] [SCORE s] [SCORE_FIELD f] [TEMPORARY secs] [STOPWORDS n ...]
/// SCHEMA field...`: build an index over the documents `schema` describes.
pub async fn Client::ft_create(
  self : Client,
  index : String,
  schema : Schema,
) -> Unit raise RedisError {
  let args : Array[Bytes] = [b"FT.CREATE", arg_string(index)]
  args.push(b"ON")
  args.push(
    match schema.on {
      Hash => b"HASH"
      Json => b"JSON"
    },
  )
  if schema.prefixes.length() > 0 {
    args.push(b"PREFIX")
    args.push(arg_int(schema.prefixes.length()))
    for prefix in schema.prefixes {
      args.push(arg_string(prefix))
    }
  }
  push_some(args, b"FILTER", schema.filter.map(arg_string))
  push_some(args, b"LANGUAGE", schema.language.map(arg_string))
  push_some(args, b"LANGUAGE_FIELD", schema.language_field.map(arg_string))
  push_some(args, b"SCORE", schema.score.map(arg_double))
  push_some(args, b"SCORE_FIELD", schema.score_field.map(arg_string))
  push_some(args, b"TEMPORARY", schema.temporary.map(arg_int64))
  match schema.stopwords {
    Some(words) => {
      args.push(b"STOPWORDS")
      args.push(arg_int(words.length()))
      for word in words {
        args.push(arg_string(word))
      }
    }
    None => ()
  }
  args.push(b"SCHEMA")
  for field in schema.fields {
    encode_schema_field(args, field)
  }
  status_ok(self.command(args))
}

///|
/// `FT.ALTER index [SKIPINITIALSCAN] SCHEMA ADD field...`: add fields to a live index.
/// `skip_initial_scan` leaves the documents already indexed unbackfilled.
pub async fn Client::ft_alter(
  self : Client,
  index : String,
  fields : Array[SchemaField],
  skip_initial_scan? : Bool,
) -> Unit raise RedisError {
  let args : Array[Bytes] = [b"FT.ALTER", arg_string(index)]
  if skip_initial_scan.unwrap_or(false) {
    args.push(b"SKIPINITIALSCAN")
  }
  args.push(b"SCHEMA")
  args.push(b"ADD")
  for field in fields {
    encode_schema_field(args, field)
  }
  status_ok(self.command(args))
}

///|
/// `FT.DROPINDEX index [DD]`: drop an index, and with `delete_documents` the documents it
/// indexed.
pub async fn Client::ft_dropindex(
  self : Client,
  index : String,
  delete_documents? : Bool,
) -> Unit raise RedisError {
  let args : Array[Bytes] = [b"FT.DROPINDEX", arg_string(index)]
  if delete_documents.unwrap_or(false) {
    args.push(b"DD")
  }
  status_ok(self.command(args))
}

///|
/// `FT._LIST`: the names of the indexes.
pub async fn Client::ft_list(self : Client) -> Array[String] raise RedisError {
  expect_text_list(self.command([b"FT._LIST"]))
}

///|
/// `FT.ALIASADD alias index`: make `alias` another name for `index`.
pub async fn Client::ft_aliasadd(
  self : Client,
  alias_name : String,
  index : String,
) -> Unit raise RedisError {
  status_ok(
    self.command([b"FT.ALIASADD", arg_string(alias_name), arg_string(index)]),
  )
}

///|
/// `FT.ALIASUPDATE alias index`: point an existing alias at another index.
pub async fn Client::ft_aliasupdate(
  self : Client,
  alias_name : String,
  index : String,
) -> Unit raise RedisError {
  status_ok(
    self.command([b"FT.ALIASUPDATE", arg_string(alias_name), arg_string(index)]),
  )
}

///|
/// `FT.ALIASDEL alias`: drop an alias.
pub async fn Client::ft_aliasdel(
  self : Client,
  alias_name : String,
) -> Unit raise RedisError {
  status_ok(self.command([b"FT.ALIASDEL", arg_string(alias_name)]))
}

///|
/// Append `PARAMS` and `DIALECT`, which the query commands take last.
fn push_query_tail(
  args : Array[Bytes],
  params : Array[(String, Bytes)],
  dialect : Int32?,
) -> Unit {
  if params.length() > 0 {
    args.push(b"PARAMS")
    args.push(arg_int(params.length() * 2))
    for (name, value) in params {
      args.push(arg_string(name))
      args.push(value)
    }
  }
  push_some(args, b"DIALECT", dialect.map(arg_int))
}

///|
/// `FT.SEARCH index query [options]`: the documents that match `query`, with the scoring,
/// paging, sorting and projection `options` asks for.
pub async fn Client::ft_search(
  self : Client,
  index : String,
  query : String,
  options? : SearchOptions,
) -> SearchResult raise RedisError {
  let options = options.unwrap_or(search_default)
  let args : Array[Bytes] = [b"FT.SEARCH", arg_string(index), arg_string(query)]
  if options.nocontent {
    args.push(b"NOCONTENT")
  }
  if options.verbatim {
    args.push(b"VERBATIM")
  }
  if options.with_scores {
    args.push(b"WITHSCORES")
  }
  if options.with_payloads {
    args.push(b"WITHPAYLOADS")
  }
  if options.with_sort_keys {
    args.push(b"WITHSORTKEYS")
  }
  if options.return_fields.length() > 0 {
    args.push(b"RETURN")
    args.push(arg_int(options.return_fields.length()))
    for field in options.return_fields {
      args.push(arg_string(field))
    }
  }
  match options.summarize {
    Some(summarize) => {
      args.push(b"SUMMARIZE")
      if summarize.fields.length() > 0 {
        args.push(arg_int(summarize.fields.length()))
        for field in summarize.fields {
          args.push(arg_string(field))
        }
      }
      push_some(args, b"FRAGS", summarize.fragments.map(arg_int64))
      push_some(args, b"LEN", summarize.length.map(arg_int64))
      push_some(args, b"SEPARATOR", summarize.separator.map(arg_string))
    }
    None => ()
  }
  if options.highlight.length() > 0 {
    args.push(b"HIGHLIGHT")
    args.push(arg_int(options.highlight.length()))
    for field in options.highlight {
      args.push(arg_string(field))
    }
  }
  push_some(args, b"SLOP", options.slop.map(arg_int64))
  push_some(args, b"TIMEOUT", options.timeout.map(arg_int64))
  if options.in_keys.length() > 0 {
    args.push(b"INKEYS")
    args.push(arg_int(options.in_keys.length()))
    for key in options.in_keys {
      args.push(arg_string(key))
    }
  }
  if options.in_fields.length() > 0 {
    args.push(b"INFIELDS")
    args.push(arg_int(options.in_fields.length()))
    for field in options.in_fields {
      args.push(arg_string(field))
    }
  }
  match options.sort_by {
    Some(sort) => {
      args.push(b"SORTBY")
      args.push(arg_string(sort.field))
      args.push(if sort.ascending { b"ASC" } else { b"DESC" })
    }
    None => ()
  }
  match options.filter {
    Some(filter) => {
      args.push(b"FILTER")
      args.push(arg_string(filter.field))
      args.push(arg_string(filter.min))
      args.push(arg_string(filter.max))
    }
    None => ()
  }
  match options.geo_filter {
    Some(filter) => {
      args.push(b"GEOFILTER")
      args.push(arg_string(filter.field))
      args.push(arg_double(filter.longitude))
      args.push(arg_double(filter.latitude))
      args.push(arg_double(filter.radius))
      args.push(arg_string(filter.unit))
    }
    None => ()
  }
  if options.limit > 0 {
    args.push(b"LIMIT")
    args.push(arg_int64(options.offset))
    args.push(arg_int64(options.limit))
  }
  push_some(args, b"SCORER", options.scorer.map(arg_string))
  push_some(args, b"LANGUAGE", options.language.map(arg_string))
  if options.explain_score {
    args.push(b"EXPLAINSCORE")
  }
  push_query_tail(args, options.params, options.dialect)
  decode_search(self.command(args), options)
}

///|
/// The extras a search reply carries between the id and the fields, in the order the
/// server writes them.
priv struct HitShape {
  with_scores : Bool
  with_payloads : Bool
  with_sort_keys : Bool
  nocontent : Bool
}

///|
/// A search reply as a `SearchResult`, from either protocol.
fn decode_search(
  reply : @resp.RespValue,
  options : SearchOptions,
) -> SearchResult raise RedisError {
  let shape : HitShape = {
    with_scores: options.with_scores,
    with_payloads: options.with_payloads,
    with_sort_keys: options.with_sort_keys,
    nocontent: options.nocontent,
  }
  match reply {
    @resp.RespMap(_) => decode_search_resp3(reply)
    @resp.Array(items) => decode_search_resp2(items, shape)
    other =>
      raise Protocol("expected a search reply, got " + @resp.resp_kind(other))
  }
}

///|
/// RESP2: `[total, id, score?, payload?, sortkey?, [field, value...], ...]`.
fn decode_search_resp2(
  items : Array[@resp.RespValue],
  shape : HitShape,
) -> SearchResult raise RedisError {
  guard items.length() > 0 else { raise Protocol("empty search reply") }
  let total = expect_int64(items[0])
  let hits : Array[SearchHit] = []
  let mut at = 1
  while at < items.length() {
    let id = expect_text(items[at])
    at += 1
    let score = if shape.with_scores {
      let value = expect_double(items[at])
      at += 1
      Some(value)
    } else {
      None
    }
    let payload = if shape.with_payloads {
      let value = expect_bytes(items[at])
      at += 1
      value
    } else {
      None
    }
    let sort_key = if shape.with_sort_keys {
      let value = expect_text(items[at])
      at += 1
      Some(value)
    } else {
      None
    }
    let fields : Array[(String, @resp.RespValue)] = []
    if !shape.nocontent && at < items.length() {
      let flat = expect_array(items[at])
      at += 1
      let mut index = 0
      while index + 1 < flat.length() {
        fields.push((expect_text(flat[index]), flat[index + 1]))
        index += 2
      }
    }
    hits.push({ id, fields, score, payload, sort_key, })
  }
  { total, hits, warnings: [], }
}

///|
/// RESP3: a map with `total_results`, `results` (each a map of `id`, `score`, `payload`,
/// `sortkey` and `extra_attributes`) and `warning`.
fn decode_search_resp3(
  reply : @resp.RespValue,
) -> SearchResult raise RedisError {
  let mut total = 0L
  let hits : Array[SearchHit] = []
  let warnings : Array[String] = []
  for (key, value) in expect_map(reply) {
    match expect_text(key) {
      "total_results" => total = expect_int64(value)
      "warning" =>
        for item in expect_array(value) {
          warnings.push(expect_text(item))
        }
      "results" =>
        for item in expect_array(value) {
          hits.push(decode_hit_resp3(item))
        }
      _ => ()
    }
  }
  { total, hits, warnings, }
}

///|
/// One RESP3 result entry as a `SearchHit`.
fn decode_hit_resp3(reply : @resp.RespValue) -> SearchHit raise RedisError {
  let mut id = ""
  let mut score : Double? = None
  let mut payload : Bytes? = None
  let mut sort_key : String? = None
  let fields : Array[(String, @resp.RespValue)] = []
  for (key, value) in expect_map(reply) {
    match expect_text(key) {
      "id" => id = expect_text(value)
      "score" => score = Some(expect_double(value))
      "payload" => payload = expect_bytes(value)
      "sortkey" => sort_key = Some(expect_text(value))
      "extra_attributes" =>
        for (name, field) in expect_map(value) {
          fields.push((expect_text(name), field))
        }
      _ => ()
    }
  }
  { id, fields, score, payload, sort_key, }
}

///|
/// `FT.AGGREGATE index query [options]`: run the pipeline of loads, groupings, applies,
/// sorts and filters `options` describes over the matching documents.
pub async fn Client::ft_aggregate(
  self : Client,
  index : String,
  query : String,
  options? : AggregateOptions,
) -> AggregateResult raise RedisError {
  let options = options.unwrap_or(aggregate_default)
  let args : Array[Bytes] = [
    b"FT.AGGREGATE",
    arg_string(index),
    arg_string(query),
  ]
  if options.verbatim {
    args.push(b"VERBATIM")
  }
  if options.load.length() > 0 {
    args.push(b"LOAD")
    args.push(arg_int(options.load.length()))
    for field in options.load {
      args.push(arg_string(field))
    }
  }
  if options.group_by.length() > 0 {
    args.push(b"GROUPBY")
    args.push(arg_int(options.group_by.length()))
    for field in options.group_by {
      args.push(arg_string(field))
    }
    for reducer in options.reducers {
      args.push(b"REDUCE")
      args.push(arg_string(reducer.function))
      args.push(arg_int(reducer.args.length()))
      for arg in reducer.args {
        args.push(arg_string(arg))
      }
      push_some(args, b"AS", reducer.as_.map(arg_string))
    }
  }
  for (expression, name) in options.apply {
    args.push(b"APPLY")
    args.push(arg_string(expression))
    args.push(b"AS")
    args.push(arg_string(name))
  }
  if options.sort_by.length() > 0 {
    args.push(b"SORTBY")
    args.push(arg_int(options.sort_by.length() * 2))
    for sort in options.sort_by {
      args.push(arg_string(sort.field))
      args.push(if sort.ascending { b"ASC" } else { b"DESC" })
    }
  }
  if options.limit > 0 {
    args.push(b"LIMIT")
    args.push(arg_int64(options.offset))
    args.push(arg_int64(options.limit))
  }
  push_some(args, b"FILTER", options.filter.map(arg_string))
  push_some(args, b"TIMEOUT", options.timeout.map(arg_int64))
  match options.cursor_count {
    Some(count) => {
      args.push(b"WITHCURSOR")
      args.push(b"COUNT")
      args.push(arg_int64(count))
      push_some(args, b"MAXIDLE", options.max_idle.map(arg_int64))
    }
    None => ()
  }
  push_query_tail(args, options.params, options.dialect)
  decode_aggregate(self.command(args), options.cursor_count is Some(_))
}

///|
/// `FT.CURSOR READ index cursor [COUNT n]`: the next page of an aggregation.
pub async fn Client::ft_cursor_read(
  self : Client,
  index : String,
  cursor : Int64,
  count? : Int64,
) -> AggregateResult raise RedisError {
  let args : Array[Bytes] = [
    b"FT.CURSOR",
    b"READ",
    arg_string(index),
    arg_int64(cursor),
  ]
  push_some(args, b"COUNT", count.map(arg_int64))
  decode_aggregate(self.command(args), true)
}

///|
/// `FT.CURSOR DEL index cursor`: drop a cursor and the rows it still holds.
pub async fn Client::ft_cursor_del(
  self : Client,
  index : String,
  cursor : Int64,
) -> Unit raise RedisError {
  status_ok(
    self.command([b"FT.CURSOR", b"DEL", arg_string(index), arg_int64(cursor)]),
  )
}

///|
/// An aggregation reply as an `AggregateResult`. Both protocols wrap the rows and the
/// cursor id in a two-element array when the query asked for a cursor.
fn decode_aggregate(
  reply : @resp.RespValue,
  with_cursor : Bool,
) -> AggregateResult raise RedisError {
  match reply {
    @resp.Array(items) => {
      let wrapped = with_cursor &&
        items.length() == 2 &&
        (items[0] is @resp.Array(_) || items[0] is @resp.RespMap(_))
      if wrapped {
        let body = decode_aggregate_body(items[0])
        {
          total: body.total,
          rows: body.rows,
          cursor: Some(expect_int64(items[1])),
        }
      } else {
        decode_aggregate_body(reply)
      }
    }
    other => decode_aggregate_body(other)
  }
}

///|
/// The rows of an aggregation reply, without the cursor wrapper.
fn decode_aggregate_body(
  reply : @resp.RespValue,
) -> AggregateResult raise RedisError {
  match reply {
    @resp.RespMap(_) => {
      let mut total = 0L
      let rows : Array[Array[(String, FieldValue)]] = []
      for (key, value) in expect_map(reply) {
        match expect_text(key) {
          "total_results" => total = expect_int64(value)
          "results" =>
            for item in expect_array(value) {
              rows.push(decode_row_resp3(item))
            }
          _ => ()
        }
      }
      { total, rows, cursor: None, }
    }
    @resp.Array(items) => {
      guard items.length() > 0 else { raise Protocol("empty aggregate reply") }
      {
        total: expect_int64(items[0]),
        rows: decode_rows_resp2(items[1:].to_owned()),
        cursor: None,
      }
    }
    other =>
      raise Protocol(
        "expected an aggregate reply, got " + @resp.resp_kind(other),
      )
  }
}

///|
/// RESP2 rows: each a flat array of field, value, field, value.
fn decode_rows_resp2(
  items : Array[@resp.RespValue],
) -> Array[Array[(String, FieldValue)]] raise RedisError {
  let rows : Array[Array[(String, FieldValue)]] = []
  for item in items {
    let flat = expect_array(item)
    let row : Array[(String, FieldValue)] = []
    let mut at = 0
    while at + 1 < flat.length() {
      row.push((expect_text(flat[at]), field_value(flat[at + 1])))
      at += 2
    }
    rows.push(row)
  }
  rows
}

///|
/// One RESP3 row, whose values live under `extra_attributes`.
fn decode_row_resp3(
  item : @resp.RespValue,
) -> Array[(String, FieldValue)] raise RedisError {
  let row : Array[(String, FieldValue)] = []
  for (key, value) in expect_map(item) {
    if expect_text(key) == "extra_attributes" {
      for (name, field) in expect_map(value) {
        row.push((expect_text(name), field_value(field)))
      }
    }
  }
  row
}

///|
/// `FT.TAGVALS index field`: the distinct values of a tag field.
pub async fn Client::ft_tagvals(
  self : Client,
  index : String,
  field : String,
) -> Array[String] raise RedisError {
  expect_text_list(
    self.command([b"FT.TAGVALS", arg_string(index), arg_string(field)]),
  )
}

///|
/// One term the index does not know, and what it suggests instead.
pub(all) struct SpellCheckTerm {
  term : String
  /// Each suggestion with its distance from the query term.
  suggestions : Array[(Double, String)]
}

///|
/// `FT.SPELLCHECK index query [DISTANCE d] [DIALECT d]`: the query terms the index knows
/// nothing about, and what it suggests instead.
pub async fn Client::ft_spellcheck(
  self : Client,
  index : String,
  query : String,
  distance? : Int32,
  dialect? : Int32,
) -> Array[SpellCheckTerm] raise RedisError {
  let args : Array[Bytes] = [
    b"FT.SPELLCHECK",
    arg_string(index),
    arg_string(query),
  ]
  push_some(args, b"DISTANCE", distance.map(arg_int))
  push_some(args, b"DIALECT", dialect.map(arg_int))
  // RESP3 keys the terms by name under `results`; RESP2 writes `[TERM, term, [...]]`.
  let terms : Array[(String, @resp.RespValue)] = []
  match self.command(args) {
    @resp.RespMap(map) =>
      for (key, value) in map {
        if expect_text(key) == "results" {
          for (name, suggestions) in expect_map(value) {
            terms.push((expect_text(name), suggestions))
          }
        }
      }
    other =>
      for entry in expect_array(other) {
        let fields = expect_array(entry)
        guard fields.length() >= 3 else {
          raise Protocol("unexpected FT.SPELLCHECK entry")
        }
        terms.push((expect_text(fields[1]), fields[2]))
      }
  }
  let out : Array[SpellCheckTerm] = []
  for (term, suggestions) in terms {
    let found : Array[(Double, String)] = []
    for item in expect_array(suggestions) {
      match item {
        @resp.Array(pair) => {
          guard pair.length() >= 2 else {
            raise Protocol("unexpected FT.SPELLCHECK suggestion")
          }
          found.push((expect_double(pair[0]), expect_text(pair[1])))
        }
        // RESP3 writes one `{suggestion: distance}` map per suggestion.
        @resp.RespMap(one) =>
          for (suggestion, distance) in one {
            found.push((expect_double(distance), expect_text(suggestion)))
          }
        _ => raise Protocol("unexpected FT.SPELLCHECK suggestion")
      }
    }
    out.push({ term, suggestions: found, })
  }
  out
}

///|
/// `FT.DICTADD dictionary term [term ...]`: add terms to a spelling dictionary, returning
/// how many were new.
pub async fn Client::ft_dictadd(
  self : Client,
  dictionary : String,
  terms : Array[String],
) -> Int64 raise RedisError {
  let args : Array[Bytes] = [b"FT.DICTADD", arg_string(dictionary)]
  for term in terms {
    args.push(arg_string(term))
  }
  expect_int64(self.command(args))
}

///|
/// `FT.DICTDEL dictionary term [term ...]`: remove terms, returning how many existed.
pub async fn Client::ft_dictdel(
  self : Client,
  dictionary : String,
  terms : Array[String],
) -> Int64 raise RedisError {
  let args : Array[Bytes] = [b"FT.DICTDEL", arg_string(dictionary)]
  for term in terms {
    args.push(arg_string(term))
  }
  expect_int64(self.command(args))
}

///|
/// `FT.DICTDUMP dictionary`: every term in a dictionary.
pub async fn Client::ft_dictdump(
  self : Client,
  dictionary : String,
) -> Array[String] raise RedisError {
  expect_text_list(self.command([b"FT.DICTDUMP", arg_string(dictionary)]))
}

///|
/// `FT.SYNUPDATE index group [SKIPINITIALSCAN] term [term ...]`: put the terms in one
/// synonym group.
pub async fn Client::ft_synupdate(
  self : Client,
  index : String,
  group : String,
  terms : Array[String],
  skip_initial_scan? : Bool,
) -> Unit raise RedisError {
  let args : Array[Bytes] = [
    b"FT.SYNUPDATE",
    arg_string(index),
    arg_string(group),
  ]
  if skip_initial_scan.unwrap_or(false) {
    args.push(b"SKIPINITIALSCAN")
  }
  for term in terms {
    args.push(arg_string(term))
  }
  status_ok(self.command(args))
}

///|
/// `FT.SYNDUMP index`: each term that has synonyms, and the groups it belongs to.
pub async fn Client::ft_syndump(
  self : Client,
  index : String,
) -> Array[(String, Array[String])] raise RedisError {
  let out : Array[(String, Array[String])] = []
  match self.command([b"FT.SYNDUMP", arg_string(index)]) {
    @resp.RespMap(pairs) =>
      for (term, groups) in pairs {
        out.push((expect_text(term), expect_text_list(groups)))
      }
    other => {
      let flat = expect_array(other)
      guard flat.length() % 2 == 0 else {
        raise Protocol("unexpected FT.SYNDUMP reply")
      }
      for at in 0..<(flat.length() / 2) {
        out.push(
          (expect_text(flat[at * 2]), expect_text_list(flat[at * 2 + 1])),
        )
      }
    }
  }
  out
}

///|
/// `FT.SUGADD key suggestion score [INCR] [PAYLOAD payload]`: add a completion, returning
/// how many suggestions the key holds.
pub async fn Client::ft_sugadd(
  self : Client,
  key : String,
  suggestion : String,
  score : Double,
  increment? : Bool,
  payload? : String,
) -> Int64 raise RedisError {
  let args : Array[Bytes] = [
    b"FT.SUGADD",
    arg_string(key),
    arg_string(suggestion),
    arg_double(score),
  ]
  if increment.unwrap_or(false) {
    args.push(b"INCR")
  }
  push_some(args, b"PAYLOAD", payload.map(arg_string))
  expect_int64(self.command(args))
}

///|
/// `FT.SUGGET key prefix [FUZZY] [MAX n]`: the completions starting with `prefix`.
pub async fn Client::ft_sugget(
  self : Client,
  key : String,
  prefix : String,
  fuzzy? : Bool,
  max? : Int64,
) -> Array[String] raise RedisError {
  let args : Array[Bytes] = [b"FT.SUGGET", arg_string(key), arg_string(prefix)]
  if fuzzy.unwrap_or(false) {
    args.push(b"FUZZY")
  }
  push_some(args, b"MAX", max.map(arg_int64))
  expect_text_list(self.command(args))
}

///|
/// `FT.SUGGET key prefix WITHSCORES`: the completions and their scores.
pub async fn Client::ft_sugget_scored(
  self : Client,
  key : String,
  prefix : String,
  fuzzy? : Bool,
  max? : Int64,
) -> Array[(String, Double)] raise RedisError {
  let args : Array[Bytes] = [
    b"FT.SUGGET",
    arg_string(key),
    arg_string(prefix),
    b"WITHSCORES",
  ]
  if fuzzy.unwrap_or(false) {
    args.push(b"FUZZY")
  }
  push_some(args, b"MAX", max.map(arg_int64))
  let flat = expect_array(self.command(args))
  let out : Array[(String, Double)] = []
  let mut at = 0
  while at + 1 < flat.length() {
    out.push((expect_text(flat[at]), expect_double(flat[at + 1])))
    at += 2
  }
  out
}

///|
/// `FT.SUGLEN key`: how many completions a key holds.
pub async fn Client::ft_suglen(
  self : Client,
  key : String,
) -> Int64 raise RedisError {
  expect_int64(self.command([b"FT.SUGLEN", arg_string(key)]))
}

///|
/// `FT.SUGDEL key suggestion`: remove a completion. True if it was there.
pub async fn Client::ft_sugdel(
  self : Client,
  key : String,
  suggestion : String,
) -> Bool raise RedisError {
  expect_bool(
    self.command([b"FT.SUGDEL", arg_string(key), arg_string(suggestion)]),
  )
}

///|
/// `FT.CONFIG GET pattern`: the settings whose names match.
pub async fn Client::ft_config_get(
  self : Client,
  pattern : String,
) -> Array[(String, String)] raise RedisError {
  let out : Array[(String, String)] = []
  match self.command([b"FT.CONFIG", b"GET", arg_string(pattern)]) {
    @resp.RespMap(pairs) =>
      for (name, value) in pairs {
        out.push((expect_text(name), expect_text(value)))
      }
    other =>
      for entry in expect_array(other) {
        let pair = expect_array(entry)
        guard pair.length() >= 2 else {
          raise Protocol("unexpected FT.CONFIG GET entry")
        }
        out.push((expect_text(pair[0]), expect_text(pair[1])))
      }
  }
  out
}

///|
/// `FT.CONFIG SET name value`: change a search setting.
pub async fn Client::ft_config_set(
  self : Client,
  name : String,
  value : String,
) -> Unit raise RedisError {
  status_ok(
    self.command([b"FT.CONFIG", b"SET", arg_string(name), arg_string(value)]),
  )
}

///|
/// `FT.EXPLAIN index query [DIALECT d]`: the query as the engine interprets it.
pub async fn Client::ft_explain(
  self : Client,
  index : String,
  query : String,
  dialect? : Int32,
) -> String raise RedisError {
  let args : Array[Bytes] = [
    b"FT.EXPLAIN",
    arg_string(index),
    arg_string(query),
  ]
  push_some(args, b"DIALECT", dialect.map(arg_int))
  expect_text(self.command(args))
}

///|
/// `FT.EXPLAINCLI index query [DIALECT d]`: the same, one line per node.
pub async fn Client::ft_explaincli(
  self : Client,
  index : String,
  query : String,
  dialect? : Int32,
) -> Array[String] raise RedisError {
  let args : Array[Bytes] = [
    b"FT.EXPLAINCLI",
    arg_string(index),
    arg_string(query),
  ]
  push_some(args, b"DIALECT", dialect.map(arg_int))
  expect_text_list(self.command(args))
}

///|
/// `FT.PROFILE index SEARCH|AGGREGATE [LIMITED] QUERY query`: the ordinary reply and the
/// timing breakdown beside it, as the protocol carries them. The second element is a
/// nested profile whose shape the search engine documents, so it stays as it arrives.
pub async fn Client::ft_profile(
  self : Client,
  index : String,
  aggregate : Bool,
  query : String,
  limited? : Bool,
) -> @resp.RespValue raise RedisError {
  let args : Array[Bytes] = [b"FT.PROFILE", arg_string(index)]
  args.push(if aggregate { b"AGGREGATE" } else { b"SEARCH" })
  if limited.unwrap_or(false) {
    args.push(b"LIMITED")
  }
  args.push(b"QUERY")
  args.push(arg_string(query))
  self.command(args)
}

///|
/// One indexed field as `FT.INFO` describes it.
pub(all) struct IndexField {
  /// The hash field or JSONPath.
  identifier : String
  /// The name queries use.
  attribute : String
  /// `TEXT`, `TAG`, `NUMERIC`, `GEO` or `VECTOR`.
  type_ : String
  sortable : Bool
  unf : Bool
  noindex : Bool
  weight : Double
}

///|
/// What `FT.INFO` reports about an index: its size, its schema and its settings.
pub(all) struct IndexInfo {
  name : String
  /// `HASH` or `JSON`.
  key_type : String
  prefixes : Array[String]
  filter : String?
  default_score : Double
  num_docs : Int64
  max_doc_id : Int64
  num_terms : Int64
  num_records : Int64
  /// Whether a background scan is still running.
  indexing : Bool
  percent_indexed : Double
  indexing_failures : Int64
  attributes : Array[IndexField]
}

///|
/// `FT.INFO index`: the index's size, schema and settings.
pub async fn Client::ft_info(
  self : Client,
  index : String,
) -> IndexInfo raise RedisError {
  let mut name = index
  let mut key_type = "HASH"
  let mut prefixes : Array[String] = []
  let mut filter : String? = None
  let mut default_score = 1.0
  let mut num_docs = 0L
  let mut max_doc_id = 0L
  let mut num_terms = 0L
  let mut num_records = 0L
  let mut indexing = false
  let mut percent_indexed = 0.0
  let mut indexing_failures = 0L
  let attributes : Array[IndexField] = []
  for (key, value) in field_pairs(self.command([b"FT.INFO", arg_string(index)])) {
    match key {
      "index_name" => name = expect_text(value)
      "num_docs" => num_docs = expect_int64(value)
      "max_doc_id" => max_doc_id = expect_int64(value)
      "num_terms" => num_terms = expect_int64(value)
      "num_records" => num_records = expect_int64(value)
      "indexing" => indexing = expect_int64(value) != 0
      "percent_indexed" => percent_indexed = expect_double(value)
      "hash_indexing_failures" => indexing_failures = expect_int64(value)
      "index_definition" =>
        for (definition_key, definition) in field_pairs(value) {
          match definition_key {
            "key_type" => key_type = expect_text(definition)
            "prefixes" => prefixes = expect_text_list(definition)
            "filter" => filter = Some(expect_text(definition))
            "default_score" => default_score = expect_double(definition)
            _ => ()
          }
        }
      "attributes" =>
        for field in expect_array(value) {
          attributes.push(decode_index_field(field))
        }
      _ => ()
    }
  }
  {
    name,
    key_type,
    prefixes,
    filter,
    default_score,
    num_docs,
    max_doc_id,
    num_terms,
    num_records,
    indexing,
    percent_indexed,
    indexing_failures,
    attributes,
  }
}

///|
/// One entry of `FT.INFO`'s `attributes`. RESP2 writes a flat list whose flags stand alone
/// (`SORTABLE`, then `UNF`); RESP3 writes a map with a `flags` array.
fn decode_index_field(reply : @resp.RespValue) -> IndexField raise RedisError {
  let mut identifier = ""
  let mut attribute = ""
  let mut type_ = ""
  let mut sortable = false
  let mut unf = false
  let mut noindex = false
  let mut weight = 1.0
  match reply {
    @resp.RespMap(pairs) =>
      for (key, value) in pairs {
        match expect_text(key) {
          "identifier" => identifier = expect_text(value)
          "attribute" => attribute = expect_text(value)
          "type" => type_ = expect_text(value)
          "WEIGHT" => weight = expect_double(value)
          "flags" =>
            for flag in expect_array(value) {
              match expect_text(flag) {
                "SORTABLE" => sortable = true
                "UNF" => unf = true
                "NOINDEX" => noindex = true
                _ => ()
              }
            }
          _ => ()
        }
      }
    other => {
      let items = expect_array(other)
      let mut at = 0
      while at < items.length() {
        let key = expect_text(items[at])
        match key {
          "SORTABLE" => {
            sortable = true
            at += 1
            if at < items.length() && expect_text(items[at]) == "UNF" {
              unf = true
              at += 1
            }
          }
          "UNF" => {
            unf = true
            at += 1
          }
          "NOINDEX" => {
            noindex = true
            at += 1
          }
          "NOSTEM"
          | "CASESENSITIVE"
          | "WITHSUFFIXTRIE"
          | "INDEXEMPTY"
          | "INDEXMISSING" => at += 1
          _ => {
            let value = if at + 1 < items.length() {
              items[at + 1]
            } else {
              @resp.Null
            }
            match key {
              "identifier" => identifier = expect_text(value)
              "attribute" => attribute = expect_text(value)
              "type" => type_ = expect_text(value)
              "WEIGHT" => weight = expect_double(value)
              _ => ()
            }
            at += 2
          }
        }
      }
    }
  }
  { identifier, attribute, type_, sortable, unf, noindex, weight, }
}

///|
/// How a hybrid query fuses the two branches.
#derive(Eq, Debug)
pub(all) enum Fusion {
  /// Reciprocal rank fusion, the default.
  Rrf
  /// A weighted sum of the two scores.
  Linear
}

///|
pub extend Fusion with Debug::{to_repr}

///|
pub extend Fusion with Eq::{not_equal, equal}

///|
/// The options `FT.HYBRID` takes besides the index, the text query and the vector.
pub(all) struct HybridOptions {
  /// How many nearest neighbours the vector branch considers.
  knn : Int64
  /// How hard the HNSW graph is searched; `None` leaves the server default.
  ef_runtime : Int64?
  /// Keep neighbours within this radius instead of the `knn` nearest, with this epsilon.
  radius : (Double, Double)?
  /// The fusion method; `None` leaves the server's `RRF`.
  fusion : Fusion?
  /// `RRF`'s (window, constant), or `LINEAR`'s (alpha, beta).
  fusion_params : (Double, Double)?
  /// The fields to return; `["*"]` returns the whole document.
  load : Array[String]
  sort_by : SortKey?
  offset : Int64
  /// 0 leaves the server's default of 10.
  limit : Int64
  /// A post-fusion filter expression.
  filter : String?
  /// Extra `PARAMS` values; the vector itself is added by `ft_hybrid`.
  params : Array[(String, Bytes)]
  timeout : Int64?
}

///|
/// `FT.HYBRID`'s options at their documented defaults.
pub let hybrid_default : HybridOptions = HybridOptions::{
  knn: 10,
  ef_runtime: None,
  radius: None,
  fusion: None,
  fusion_params: None,
  load: [],
  sort_by: None,
  offset: 0,
  limit: 0,
  filter: None,
  params: [],
  timeout: None,
}

///|
/// `FT.HYBRID index SEARCH query VSIM @field $vector [KNN ...] [COMBINE ...] [LOAD ...]
/// [SORTBY ...] [LIMIT ...] [FILTER ...] PARAMS ...`: rank documents by a text search and a
/// vector search at once. `vector` is the query vector as the field's element type writes
/// it, and `vector_field` the name of the schema field it is compared against.
pub async fn Client::ft_hybrid(
  self : Client,
  index : String,
  query : String,
  vector_field : String,
  vector : Bytes,
  options? : HybridOptions,
) -> SearchResult raise RedisError {
  let options = options.unwrap_or(hybrid_default)
  let args : Array[Bytes] = [b"FT.HYBRID", arg_string(index)]
  args.push(b"SEARCH")
  args.push(arg_string(query))
  args.push(b"VSIM")
  args.push(arg_string("@" + vector_field))
  args.push(b"$moonredis_vector")
  match options.radius {
    Some((radius, epsilon)) => {
      args.push(b"RANGE")
      args.push(b"4")
      args.push(b"RADIUS")
      args.push(arg_double(radius))
      args.push(b"EPSILON")
      args.push(arg_double(epsilon))
    }
    None => {
      let knn : Array[Bytes] = [b"K", arg_int64(options.knn)]
      match options.ef_runtime {
        Some(n) => {
          knn.push(b"EF_RUNTIME")
          knn.push(arg_int64(n))
        }
        None => ()
      }
      args.push(b"KNN")
      args.push(arg_int(knn.length()))
      for item in knn {
        args.push(item)
      }
    }
  }
  match options.fusion {
    Some(fusion) => {
      args.push(b"COMBINE")
      args.push(
        match fusion {
          Rrf => b"RRF"
          Linear => b"LINEAR"
        },
      )
      let params : Array[Bytes] = []
      match (fusion, options.fusion_params) {
        (Rrf, Some((window, constant))) => {
          params.push(b"WINDOW")
          params.push(arg_double(window))
          params.push(b"CONSTANT")
          params.push(arg_double(constant))
        }
        (Linear, Some((alpha, beta))) => {
          params.push(b"ALPHA")
          params.push(arg_double(alpha))
          params.push(b"BETA")
          params.push(arg_double(beta))
        }
        _ => ()
      }
      args.push(arg_int(params.length()))
      for item in params {
        args.push(item)
      }
    }
    None => ()
  }
  if options.load.length() > 0 {
    args.push(b"LOAD")
    args.push(arg_int(options.load.length()))
    for field in options.load {
      args.push(arg_string(field))
    }
  }
  push_some(args, b"FILTER", options.filter.map(arg_string))
  if options.limit > 0 {
    args.push(b"LIMIT")
    args.push(arg_int64(options.offset))
    args.push(arg_int64(options.limit))
  }
  match options.sort_by {
    Some(sort) => {
      args.push(b"SORTBY")
      args.push(b"2")
      args.push(arg_string(sort.field))
      args.push(if sort.ascending { b"ASC" } else { b"DESC" })
    }
    None => ()
  }
  push_some(args, b"TIMEOUT", options.timeout.map(arg_int64))
  let params : Array[(String, Bytes)] = [("moonredis_vector", vector)]
  for (name, value) in options.params {
    params.push((name, value))
  }
  push_query_tail(args, params, None)
  decode_search(self.command(args), search_default)
}