// The time series commands. A key holds a stream of (timestamp, value) samples with
// retention, labels for filtering, and optional compaction rules that roll samples into
// another series. A sample crosses the wire as text under RESP2 and as a double under
// RESP3; both arrive as a `Sample` here.
// https://redis.io/docs/latest/commands/?group=timeseries

///|
/// One sample: when it was recorded, and what it measured.
#derive(Eq, Debug)
pub(all) struct Sample {
  timestamp : Int64
  value : Double
}

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

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

///|
/// One series of a multi-series reply: its key, its labels, and the samples the query
/// selected.
pub(all) struct Series {
  key : String
  labels : Array[(String, String)]
  samples : Array[Sample]
}

///|
/// One row of a reply pivoted by timestamp: the timestamp and one value per key, `None`
/// where that key has no sample there.
pub(all) struct PivotRow {
  timestamp : Int64
  values : Array[Double?]
}

///|
/// A point in a range query: the earliest sample the series has, the latest, or a
/// timestamp.
#derive(Eq, Debug)
pub(all) enum TimeBound {
  Min
  Max
  At(Int64)
}

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

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

///|
/// The `-`, `+` or decimal text a range query writes for this bound.
pub fn TimeBound::text(self : TimeBound) -> String {
  match self {
    Min => "-"
    Max => "+"
    At(t) => t.to_string()
  }
}

///|
/// The options a range query takes: paging, filtering, and how to bucket the samples.
pub(all) struct RangeOptions {
  /// Report the compacted value of a partial bucket at the end.
  latest : Bool
  /// Keep only these timestamps.
  filter_by_ts : Array[Int64]
  /// Keep only samples in this value range.
  filter_by_value : (Double, Double)?
  /// The most samples to return; 0 returns them all.
  count : Int64
  /// `avg`, `sum`, `min`, `max`, `range`, `count`, `first`, `last`, `std.p`, `std.s`,
  /// `var.p`, `var.s` or `twa`.
  aggregation : String?
  /// The width of each aggregation bucket, in milliseconds.
  bucket : Int64
  /// Where buckets start: a timestamp, `-` (the range start) or `+` (the range end).
  align : String?
  /// How a bucket is stamped: `-`, `+` or `~`.
  bucket_timestamp : String?
  /// Report buckets with no samples as well.
  empty : Bool
}

///|
/// Range options at their documented defaults.
pub let range_default : RangeOptions = RangeOptions::{
  latest: false,
  filter_by_ts: [],
  filter_by_value: None,
  count: 0,
  aggregation: None,
  bucket: 0,
  align: None,
  bucket_timestamp: None,
  empty: false,
}

///|
/// The options a multi-series query adds: which labels to carry, and how to group.
pub(all) struct MultiRangeOptions {
  range : RangeOptions
  /// Carry each series' labels into the reply.
  with_labels : Bool
  /// Carry only these labels.
  select_labels : Array[String]
  /// Group the series by this label instead of returning each on its own.
  group_by : String?
  /// The reducer the grouped series share.
  reduce : String?
  /// Leave out a series whose sample list came back empty.
  exclude_empty : Bool
}

///|
/// Multi-series range options at their documented defaults.
pub let multi_range_default : MultiRangeOptions = MultiRangeOptions::{
  range: range_default,
  with_labels: false,
  select_labels: [],
  group_by: None,
  reduce: None,
  exclude_empty: false,
}

///|
/// The options a multi-series point query takes.
pub(all) struct MultiGetOptions {
  with_labels : Bool
  select_labels : Array[String]
  latest : Bool
}

///|
/// Multi-series get options at their documented defaults.
pub let multi_get_default : MultiGetOptions = MultiGetOptions::{
  with_labels: false,
  select_labels: [],
  latest: false,
}

///|
/// A compaction rule: the series this one rolls into, how wide each bucket is, and how the
/// bucket is reduced.
pub(all) struct CompactionRule {
  destination : String
  bucket : Int64
  aggregator : String
}

///|
/// What `TS.INFO` reports about a series.
pub(all) struct TsInfo {
  total_samples : Int64
  memory_usage : Int64
  first_timestamp : Int64
  last_timestamp : Int64
  retention_millis : Int64
  chunk_count : Int64
  chunk_size : Int64
  /// `compressed` or `uncompressed`.
  chunk_type : String
  duplicate_policy : String
  labels : Array[(String, String)]
  /// The series this one compacts from, when it is a compaction target.
  source_key : String?
  rules : Array[CompactionRule]
  ignore_max_time_diff : Int64
  ignore_max_val_diff : Double
}

///|
/// One `[timestamp, value]` pair.
fn decode_sample(reply : @resp.RespValue) -> Sample raise RedisError {
  let pair = expect_array(reply)
  guard pair.length() >= 2 else { raise Protocol("unexpected sample") }
  { timestamp: expect_int64(pair[0]), value: numeric_value(pair[1]), }
}

///|
/// A list of `[timestamp, value]` pairs.
fn decode_samples(reply : @resp.RespValue) -> Array[Sample] raise RedisError {
  expect_array(reply).map(decode_sample)
}

///|
/// The `[name, value]` pairs of a label set, from the map RESP3 sends or the array RESP2
/// sends.
fn decode_labels(
  reply : @resp.RespValue,
) -> Array[(String, String)] raise RedisError {
  let out : Array[(String, String)] = []
  match reply {
    @resp.RespMap(pairs) =>
      for (name, value) in pairs {
        out.push((expect_text(name), expect_text(value)))
      }
    other =>
      for item in expect_array(other) {
        let pair = expect_array(item)
        guard pair.length() >= 2 else { raise Protocol("unexpected label") }
        out.push((expect_text(pair[0]), expect_text(pair[1])))
      }
  }
  out
}

///|
/// Append the range options both the single and multi-series queries share.
fn push_range_options(args : Array[Bytes], options : RangeOptions) -> Unit {
  if options.latest {
    args.push(b"LATEST")
  }
  if options.filter_by_ts.length() > 0 {
    args.push(b"FILTER_BY_TS")
    for ts in options.filter_by_ts {
      args.push(arg_int64(ts))
    }
  }
  match options.filter_by_value {
    Some((min, max)) => {
      args.push(b"FILTER_BY_VALUE")
      args.push(arg_double(min))
      args.push(arg_double(max))
    }
    None => ()
  }
  if options.count > 0 {
    args.push(b"COUNT")
    args.push(arg_int64(options.count))
  }
  match options.aggregation {
    Some(aggregator) => {
      match options.align {
        Some(align) => {
          args.push(b"ALIGN")
          args.push(arg_string(align))
        }
        None => ()
      }
      args.push(b"AGGREGATION")
      args.push(arg_string(aggregator))
      args.push(arg_int64(options.bucket))
      match options.bucket_timestamp {
        Some(stamp) => {
          args.push(b"BUCKETTIMESTAMP")
          args.push(arg_string(stamp))
        }
        None => ()
      }
      if options.empty {
        args.push(b"EMPTY")
      }
    }
    None => ()
  }
}

///|
/// Append the label options a multi-series query takes.
fn push_label_options(
  args : Array[Bytes],
  with_labels : Bool,
  select_labels : Array[String],
) -> Unit {
  if select_labels.length() > 0 {
    args.push(b"SELECTED_LABELS")
    for label in select_labels {
      args.push(arg_string(label))
    }
  } else if with_labels {
    args.push(b"WITHLABELS")
  }
}

///|
/// Append the `LABELS` block `TS.CREATE` and its kin take.
fn push_ts_labels(
  args : Array[Bytes],
  labels : Array[(String, String)]?,
) -> Unit {
  match labels {
    Some(pairs) =>
      if pairs.length() > 0 {
        args.push(b"LABELS")
        for (name, value) in pairs {
          args.push(arg_string(name))
          args.push(arg_string(value))
        }
      }
    None => ()
  }
}

///|
/// `TS.CREATE key [RETENTION ms] [ENCODING COMPRESSED|UNCOMPRESSED] [CHUNK_SIZE bytes]
/// [DUPLICATE_POLICY policy] [IGNORE max_time_diff max_val_diff] [LABELS ...]`: create a
/// series, optionally dropping samples older than `retention`.
pub async fn Client::ts_create(
  self : Client,
  key : String,
  retention? : Int64,
  uncompressed? : Bool,
  chunk_size? : Int64,
  duplicate_policy? : String,
  ignore_max_time_diff? : Int64,
  ignore_max_val_diff? : Double,
  labels? : Array[(String, String)],
) -> Unit raise RedisError {
  let args : Array[Bytes] = [b"TS.CREATE", arg_string(key)]
  push_some(args, b"RETENTION", retention.map(arg_int64))
  if uncompressed.unwrap_or(false) {
    args.push(b"ENCODING")
    args.push(b"UNCOMPRESSED")
  }
  push_some(args, b"CHUNK_SIZE", chunk_size.map(arg_int64))
  push_some(args, b"DUPLICATE_POLICY", duplicate_policy.map(arg_string))
  match (ignore_max_time_diff, ignore_max_val_diff) {
    (Some(time), Some(value)) => {
      args.push(b"IGNORE")
      args.push(arg_int64(time))
      args.push(arg_double(value))
    }
    _ => ()
  }
  push_ts_labels(args, labels)
  status_ok(self.command(args))
}

///|
/// `TS.ALTER key [RETENTION ms] [CHUNK_SIZE bytes] [DUPLICATE_POLICY policy]
/// [IGNORE ...] [LABELS ...]`: change a series' settings; `LABELS` replaces the label set.
pub async fn Client::ts_alter(
  self : Client,
  key : String,
  retention? : Int64,
  chunk_size? : Int64,
  duplicate_policy? : String,
  ignore_max_time_diff? : Int64,
  ignore_max_val_diff? : Double,
  labels? : Array[(String, String)],
) -> Unit raise RedisError {
  let args : Array[Bytes] = [b"TS.ALTER", arg_string(key)]
  push_some(args, b"RETENTION", retention.map(arg_int64))
  push_some(args, b"CHUNK_SIZE", chunk_size.map(arg_int64))
  push_some(args, b"DUPLICATE_POLICY", duplicate_policy.map(arg_string))
  match (ignore_max_time_diff, ignore_max_val_diff) {
    (Some(time), Some(value)) => {
      args.push(b"IGNORE")
      args.push(arg_int64(time))
      args.push(arg_double(value))
    }
    _ => ()
  }
  push_ts_labels(args, labels)
  status_ok(self.command(args))
}

///|
/// `TS.ADD key timestamp value [RETENTION ms] [ENCODING ...] [CHUNK_SIZE bytes]
/// [ON_DUPLICATE policy] [LABELS ...]`: append one sample, returning its timestamp.
pub async fn Client::ts_add(
  self : Client,
  key : String,
  timestamp : Int64,
  value : Double,
  retention? : Int64,
  uncompressed? : Bool,
  chunk_size? : Int64,
  on_duplicate? : String,
  labels? : Array[(String, String)],
) -> Int64 raise RedisError {
  let args : Array[Bytes] = [
    b"TS.ADD",
    arg_string(key),
    arg_int64(timestamp),
    arg_double(value),
  ]
  push_some(args, b"RETENTION", retention.map(arg_int64))
  if uncompressed.unwrap_or(false) {
    args.push(b"ENCODING")
    args.push(b"UNCOMPRESSED")
  }
  push_some(args, b"CHUNK_SIZE", chunk_size.map(arg_int64))
  push_some(args, b"ON_DUPLICATE", on_duplicate.map(arg_string))
  push_ts_labels(args, labels)
  expect_int64(self.command(args))
}

///|
/// `TS.MADD key timestamp value [key timestamp value ...]`: append several samples at once,
/// returning each timestamp, `None` where that sample was refused.
pub async fn Client::ts_madd(
  self : Client,
  samples : Array[(String, Int64, Double)],
) -> Array[Int64?] raise RedisError {
  let args : Array[Bytes] = [b"TS.MADD"]
  for (key, timestamp, value) in samples {
    args.push(arg_string(key))
    args.push(arg_int64(timestamp))
    args.push(arg_double(value))
  }
  let out : Array[Int64?] = []
  for item in expect_array(self.command(args)) {
    match item {
      @resp.Error(_) | @resp.BulkError(_) => out.push(None)
      other => out.push(Some(expect_int64(other)))
    }
  }
  out
}

///|
/// `TS.INCRBY key value [TIMESTAMP ts] [RETENTION ms] [ENCODING ...] [CHUNK_SIZE bytes]
/// [DUPLICATE_POLICY policy] [LABELS ...]`: add `value` to the sample at `timestamp`,
/// creating the sample when there is none, and return the timestamp.
pub async fn Client::ts_incrby(
  self : Client,
  key : String,
  value : Double,
  timestamp? : Int64,
  retention? : Int64,
  uncompressed? : Bool,
  chunk_size? : Int64,
  duplicate_policy? : String,
  labels? : Array[(String, String)],
) -> Int64 raise RedisError {
  let args : Array[Bytes] = [b"TS.INCRBY", arg_string(key), arg_double(value)]
  push_some(args, b"TIMESTAMP", timestamp.map(arg_int64))
  push_some(args, b"RETENTION", retention.map(arg_int64))
  if uncompressed.unwrap_or(false) {
    args.push(b"ENCODING")
    args.push(b"UNCOMPRESSED")
  }
  push_some(args, b"CHUNK_SIZE", chunk_size.map(arg_int64))
  push_some(args, b"DUPLICATE_POLICY", duplicate_policy.map(arg_string))
  push_ts_labels(args, labels)
  expect_int64(self.command(args))
}

///|
/// `TS.DECRBY key value [TIMESTAMP ts] ...`: subtract `value` from the sample at
/// `timestamp`, returning the timestamp.
pub async fn Client::ts_decrby(
  self : Client,
  key : String,
  value : Double,
  timestamp? : Int64,
  retention? : Int64,
  uncompressed? : Bool,
  chunk_size? : Int64,
  duplicate_policy? : String,
  labels? : Array[(String, String)],
) -> Int64 raise RedisError {
  let args : Array[Bytes] = [b"TS.DECRBY", arg_string(key), arg_double(value)]
  push_some(args, b"TIMESTAMP", timestamp.map(arg_int64))
  push_some(args, b"RETENTION", retention.map(arg_int64))
  if uncompressed.unwrap_or(false) {
    args.push(b"ENCODING")
    args.push(b"UNCOMPRESSED")
  }
  push_some(args, b"CHUNK_SIZE", chunk_size.map(arg_int64))
  push_some(args, b"DUPLICATE_POLICY", duplicate_policy.map(arg_string))
  push_ts_labels(args, labels)
  expect_int64(self.command(args))
}

///|
/// `TS.DEL key fromTimestamp toTimestamp`: delete the samples in the range, returning how
/// many were removed.
pub async fn Client::ts_del(
  self : Client,
  key : String,
  from_ : String,
  to : String,
) -> Int64 raise RedisError {
  expect_int64(
    self.command([b"TS.DEL", arg_string(key), arg_string(from_), arg_string(to)]),
  )
}

///|
/// `TS.CREATERULE source destination AGGREGATION aggregator bucketDuration
/// [alignTimestamp]`: compact `source` into `destination` in buckets of `bucket` ms.
pub async fn Client::ts_createrule(
  self : Client,
  source : String,
  destination : String,
  aggregator : String,
  bucket : Int64,
  align_timestamp? : Int64,
) -> Unit raise RedisError {
  let args : Array[Bytes] = [
    b"TS.CREATERULE",
    arg_string(source),
    arg_string(destination),
    b"AGGREGATION",
    arg_string(aggregator),
    arg_int64(bucket),
  ]
  push_some(args, b"", align_timestamp.map(arg_int64))
  status_ok(self.command(args))
}

///|
/// `TS.DELETERULE source destination`: stop compacting `source` into `destination`.
pub async fn Client::ts_deleterule(
  self : Client,
  source : String,
  destination : String,
) -> Unit raise RedisError {
  status_ok(
    self.command([b"TS.DELETERULE", arg_string(source), arg_string(destination)]),
  )
}

///|
/// `TS.RANGE key fromTimestamp toTimestamp [options]`: the samples in the range, oldest
/// first.
pub async fn Client::ts_range(
  self : Client,
  key : String,
  from_ : TimeBound,
  to : TimeBound,
  options? : RangeOptions,
) -> Array[Sample] raise RedisError {
  let args : Array[Bytes] = [
    b"TS.RANGE",
    arg_string(key),
    arg_string(from_.text()),
    arg_string(to.text()),
  ]
  push_range_options(args, options.unwrap_or(range_default))
  decode_samples(self.command(args))
}

///|
/// `TS.REVRANGE key fromTimestamp toTimestamp [options]`: the samples in the range, newest
/// first. The bounds keep the order `TS.RANGE` uses; only the reply is reversed.
pub async fn Client::ts_revrange(
  self : Client,
  key : String,
  from_ : TimeBound,
  to : TimeBound,
  options? : RangeOptions,
) -> Array[Sample] raise RedisError {
  let args : Array[Bytes] = [
    b"TS.REVRANGE",
    arg_string(key),
    arg_string(from_.text()),
    arg_string(to.text()),
  ]
  push_range_options(args, options.unwrap_or(range_default))
  decode_samples(self.command(args))
}

///|
/// `TS.MRANGE fromTimestamp toTimestamp [options] FILTER filter...`: the samples of every
/// series whose labels match, oldest first.
pub async fn Client::ts_mrange(
  self : Client,
  from_ : TimeBound,
  to : TimeBound,
  filters : Array[String],
  options? : MultiRangeOptions,
) -> Array[Series] raise RedisError {
  let args : Array[Bytes] = [
    b"TS.MRANGE",
    arg_string(from_.text()),
    arg_string(to.text()),
  ]
  let options = options.unwrap_or(multi_range_default)
  push_multi_range_options(args, options)
  args.push(b"FILTER")
  for filter in filters {
    args.push(arg_string(filter))
  }
  decode_series(self.command(args))
}

///|
/// `TS.MREVRANGE fromTimestamp toTimestamp [options] FILTER filter...`: the same, newest
/// first. The bounds keep the order `TS.MRANGE` uses.
pub async fn Client::ts_mrevrange(
  self : Client,
  from_ : TimeBound,
  to : TimeBound,
  filters : Array[String],
  options? : MultiRangeOptions,
) -> Array[Series] raise RedisError {
  let args : Array[Bytes] = [
    b"TS.MREVRANGE",
    arg_string(from_.text()),
    arg_string(to.text()),
  ]
  let options = options.unwrap_or(multi_range_default)
  push_multi_range_options(args, options)
  args.push(b"FILTER")
  for filter in filters {
    args.push(arg_string(filter))
  }
  decode_series(self.command(args))
}

///|
/// Append what a multi-series range query takes before its `FILTER`.
fn push_multi_range_options(
  args : Array[Bytes],
  options : MultiRangeOptions,
) -> Unit {
  push_label_options(args, options.with_labels, options.select_labels)
  push_range_options(args, options.range)
  match (options.group_by, options.reduce) {
    (Some(label), Some(reducer)) => {
      args.push(b"GROUPBY")
      args.push(arg_string(label))
      args.push(b"REDUCE")
      args.push(arg_string(reducer))
    }
    _ => ()
  }
  if options.exclude_empty {
    args.push(b"EXCLUDEEMPTY")
  }
}

///|
/// `TS.MGET [options] FILTER filter...`: the latest sample of every series whose labels
/// match.
pub async fn Client::ts_mget(
  self : Client,
  filters : Array[String],
  options? : MultiGetOptions,
) -> Array[Series] raise RedisError {
  let options = options.unwrap_or(multi_get_default)
  let args : Array[Bytes] = [b"TS.MGET"]
  push_label_options(args, options.with_labels, options.select_labels)
  if options.latest {
    args.push(b"LATEST")
  }
  args.push(b"FILTER")
  for filter in filters {
    args.push(arg_string(filter))
  }
  decode_series_points(self.command(args))
}

///|
/// A multi-series range reply as `Series` entries.
fn decode_series(reply : @resp.RespValue) -> Array[Series] raise RedisError {
  decode_series_impl(reply, false)
}

///|
/// A multi-series point reply, whose samples list holds at most one sample.
fn decode_series_points(
  reply : @resp.RespValue,
) -> Array[Series] raise RedisError {
  decode_series_impl(reply, true)
}

///|
/// A multi-series reply as `Series` entries. RESP2 writes `[key, labels, samples]` per
/// series; RESP3 keys a map by name and writes `[labels, aggregators, samples]`. `single`
/// says whether the last element is one sample rather than a list of them.
fn decode_series_impl(
  reply : @resp.RespValue,
  single : Bool,
) -> Array[Series] raise RedisError {
  let out : Array[Series] = []
  match reply {
    @resp.RespMap(pairs) =>
      for (name, value) in pairs {
        let parts = expect_array(value)
        guard parts.length() >= 2 else { raise Protocol("unexpected series") }
        out.push({
          key: expect_text(name),
          labels: decode_labels(parts[0]),
          samples: decode_last_samples(parts[parts.length() - 1], single),
        })
      }
    other =>
      for item in expect_array(other) {
        let parts = expect_array(item)
        guard parts.length() >= 3 else { raise Protocol("unexpected series") }
        out.push({
          key: expect_text(parts[0]),
          labels: decode_labels(parts[1]),
          samples: decode_last_samples(parts[2], single),
        })
      }
  }
  out
}

///|
/// The samples at the end of a series: a list of them, or the one a point query returns.
fn decode_last_samples(
  reply : @resp.RespValue,
  single : Bool,
) -> Array[Sample] raise RedisError {
  if !single {
    return decode_samples(reply)
  }
  let items = expect_array(reply)
  if items.length() == 0 {
    []
  } else {
    [decode_sample(reply)]
  }
}

///|
/// `TS.GET key [LATEST]`: the newest sample, or `None` when the series is empty.
pub async fn Client::ts_get(
  self : Client,
  key : String,
  latest? : Bool,
) -> Sample? raise RedisError {
  let args : Array[Bytes] = [b"TS.GET", arg_string(key)]
  if latest.unwrap_or(false) {
    args.push(b"LATEST")
  }
  match self.command(args) {
    @resp.Null => None
    other => Some(decode_sample(other))
  }
}

///|
/// `TS.INFO key`: the series' size, settings, labels and compaction rules.
pub async fn Client::ts_info(
  self : Client,
  key : String,
) -> TsInfo raise RedisError {
  let reply = self.command([b"TS.INFO", arg_string(key)])
  let info = decode_info(reply)
  let mut labels : Array[(String, String)] = []
  let mut rules : Array[CompactionRule] = []
  // Labels and rules are nested lists, which the flat `Info` view drops, so they are read
  // from the reply itself.
  for (name, value) in field_pairs(reply) {
    match name {
      "labels" => labels = decode_labels(value)
      "rules" => rules = decode_rules(value)
      _ => ()
    }
  }
  {
    total_samples: info.int("totalSamples").unwrap_or(0),
    memory_usage: info.int("memoryUsage").unwrap_or(0),
    first_timestamp: info.int("firstTimestamp").unwrap_or(0),
    last_timestamp: info.int("lastTimestamp").unwrap_or(0),
    retention_millis: info.int("retentionTime").unwrap_or(0),
    chunk_count: info.int("chunkCount").unwrap_or(0),
    chunk_size: info.int("chunkSize").unwrap_or(0),
    chunk_type: info.text("chunkType").unwrap_or(""),
    duplicate_policy: info.text("duplicatePolicy").unwrap_or(""),
    labels,
    source_key: info.text("sourceKey"),
    rules,
    ignore_max_time_diff: info.int("ignoreMaxTimeDiff").unwrap_or(0),
    ignore_max_val_diff: info.double("ignoreMaxValDiff").unwrap_or(0.0),
  }
}

///|
/// The compaction rules of a series: a list of `[destination, bucket, aggregator, align]`
/// under RESP2, a map keyed by destination under RESP3.
fn decode_rules(
  reply : @resp.RespValue,
) -> Array[CompactionRule] raise RedisError {
  let rules : Array[CompactionRule] = []
  match reply {
    @resp.RespMap(pairs) =>
      for (destination, body) in pairs {
        let parts = expect_array(body)
        guard parts.length() >= 2 else {
          raise Protocol("unexpected compaction rule body")
        }
        rules.push({
          destination: expect_text(destination),
          bucket: expect_int64(parts[0]),
          aggregator: expect_text(parts[1]),
        })
      }
    @resp.Array(items) =>
      for item in items {
        let parts = expect_array(item)
        guard parts.length() >= 3 else {
          raise Protocol("unexpected compaction rule")
        }
        rules.push({
          destination: expect_text(parts[0]),
          bucket: expect_int64(parts[1]),
          aggregator: expect_text(parts[2]),
        })
      }
    @resp.Null => ()
    other =>
      raise Protocol("expected compaction rules, got " + @resp.resp_kind(other))
  }
  rules
}

///|
/// `TS.QUERYINDEX filter...`: the keys whose labels match every filter.
pub async fn Client::ts_queryindex(
  self : Client,
  filters : Array[String],
) -> Array[String] raise RedisError {
  let args : Array[Bytes] = [b"TS.QUERYINDEX"]
  for filter in filters {
    args.push(arg_string(filter))
  }
  expect_text_list(self.command(args))
}

///|
/// `TS.QUERYLABELS [FILTER filter...]`: every label name in use.
pub async fn Client::ts_querylabels(
  self : Client,
  filters? : Array[String],
) -> Array[String] raise RedisError {
  let args : Array[Bytes] = [b"TS.QUERYLABELS", b"LABELS"]
  match filters {
    Some(filters) =>
      if filters.length() > 0 {
        args.push(b"FILTER")
        for filter in filters {
          args.push(arg_string(filter))
        }
      }
    None => ()
  }
  expect_text_list(self.command(args))
}

///|
/// `TS.QUERYLABELS VALUES label [FILTER filter...]`: every value of one label in use.
pub async fn Client::ts_querylabel_values(
  self : Client,
  label : String,
  filters? : Array[String],
) -> Array[String] raise RedisError {
  let args : Array[Bytes] = [b"TS.QUERYLABELS", b"VALUES", arg_string(label)]
  match filters {
    Some(filters) =>
      if filters.length() > 0 {
        args.push(b"FILTER")
        for filter in filters {
          args.push(arg_string(filter))
        }
      }
    None => ()
  }
  expect_text_list(self.command(args))
}

///|
/// `TS.READ key timestamp [BLOCK ms count] [MAX_COUNT n]`: up to `max_count` samples from
/// `timestamp` on, waiting for at least `block`'s count when it is given.
pub async fn Client::ts_read(
  self : Client,
  key : String,
  timestamp : Int64,
  block? : (Int64, Int64),
  max_count? : Int64,
) -> Array[Sample] raise RedisError {
  let args : Array[Bytes] = [b"TS.READ", arg_string(key), arg_int64(timestamp)]
  match block {
    Some((milliseconds, min_count)) => {
      args.push(b"BLOCK")
      args.push(arg_int64(milliseconds))
      args.push(arg_int64(min_count))
    }
    None => ()
  }
  push_some(args, b"MAX_COUNT", max_count.map(arg_int64))
  decode_samples(self.command(args))
}

///|
/// `TS.NRANGE numkeys key... fromTimestamp toTimestamp [options]`: the samples of several
/// series pivoted by timestamp, one value column per key in the order the keys are given.
pub async fn Client::ts_nrange(
  self : Client,
  keys : Array[String],
  from_ : TimeBound,
  to : TimeBound,
  options? : RangeOptions,
) -> Array[PivotRow] raise RedisError {
  let args : Array[Bytes] = [b"TS.NRANGE", arg_int(keys.length())]
  for key in keys {
    args.push(arg_string(key))
  }
  args.push(arg_string(from_.text()))
  args.push(arg_string(to.text()))
  push_range_options(args, options.unwrap_or(range_default))
  decode_pivot(self.command(args))
}

///|
/// `TS.NREVRANGE numkeys key... fromTimestamp toTimestamp [options]`: the same, newest
/// first. The bounds keep the order `TS.NRANGE` uses.
pub async fn Client::ts_nrevrange(
  self : Client,
  keys : Array[String],
  from_ : TimeBound,
  to : TimeBound,
  options? : RangeOptions,
) -> Array[PivotRow] raise RedisError {
  let args : Array[Bytes] = [b"TS.NREVRANGE", arg_int(keys.length())]
  for key in keys {
    args.push(arg_string(key))
  }
  args.push(arg_string(from_.text()))
  args.push(arg_string(to.text()))
  push_range_options(args, options.unwrap_or(range_default))
  decode_pivot(self.command(args))
}

///|
/// A pivoted reply: `[timestamp, [value per key]]`, with `NaN` where a key has no sample.
fn decode_pivot(reply : @resp.RespValue) -> Array[PivotRow] raise RedisError {
  let out : Array[PivotRow] = []
  for item in expect_array(reply) {
    let parts = expect_array(item)
    guard parts.length() >= 2 else { raise Protocol("unexpected pivoted row") }
    let values : Array[Double?] = []
    for value in expect_array(parts[1]) {
      let sample = numeric_value(value)
      values.push(if sample.is_nan() { None } else { Some(sample) })
    }
    out.push({ timestamp: expect_int64(parts[0]), values, })
  }
  out
}