yo-resp 0.3.25

//! `TS.*`, the time series family RedisTimeSeries put on the wire.
//!
//! The series itself is in `yo-series` and this is the wire in front of it, the
//! same split as `super::cms` and the other module families. Twenty one
//! commands here: the two that shape a series, the four that write samples into
//! it, the one that cuts a span back out, the one that takes the newest sample,
//! the one that reports on it, the two that read a span back either way round,
//! the three the 8.10 release added for reading several series at once, the
//! three that search on labels, the two that read a span out of every matching
//! series and the two that set up and take down a compaction rule.
//!
//! # Errors
//!
//! Nearly all of them are one sentence starting `TSDB:` behind Redis's own
//! `ERR`, and they are copied word for word. Two are written straight to the
//! client with no prefix at all, a negative `RETENTION` and a `TS.INCRBY` whose
//! timestamp goes backwards, which is not a style anyone chose but is what a
//! client sees and so is what this writes.
//!
//! # The options are a scan and not a grammar
//!
//! Every option word is looked for across the whole command rather than read in
//! order, so a key called `RETENTION` is the `RETENTION` keyword, a word that is
//! not an option at all is ignored, and `LABELS` swallows everything after it in
//! pairs while the other keywords still find themselves inside what it
//! swallowed. All of that is visible from a client, so all of it is copied. A
//! client that writes its options in the documented order never sees any of it.
//!
//! # The order the checks happen in
//!
//! `TS.CREATE` and `TS.ALTER` read their options before they look at the key, so
//! a bad retention against a key that is already there answers about the
//! retention. `TS.ADD` is the other way round for everything except the value
//! and the timestamp, which it reads first. `TS.MADD` never creates a key and
//! never reads an option, so a missing key inside it is the key is not a TSDB
//! key rather than anything about creating one.
//!
//! `TS.RANGE` and `TS.REVRANGE` are the other way round again: the arity, then
//! the key, then the two ends of the span, then everything else. A read of a key
//! that is not there answers that it is not there even when the rest of the
//! command is nonsense. The option scan starts at the `to` slot rather than one
//! past it, which never finds anything because that slot has already had to be a
//! timestamp, but it is where the reference starts so it is where this starts.
//!
//! # What `COUNT` is allowed to be
//!
//! A `COUNT` that lands where a reduction name or a reducer name belongs is not
//! a `COUNT`, it is that name, and the scan starts again two words later. If the
//! second scan finds nothing then the read has no count at all rather than an
//! error, so `AGGREGATION count 100` counts and does not truncate.
//!
//! # What `last` carries into a bucket it has nothing for
//!
//! There are two of these and they do not behave the same way, which is worth
//! writing down because it looks like one rule until it is measured. A bucket
//! with no kept readings at all, whether the readings were never there or the
//! filters took them away, carries the reading before the gap in time, and it
//! does that whichever way the read runs. A bucket that has kept readings but
//! none `last` can use, so every one of them is not a number, carries whatever
//! the bucket before it in the reading direction answered, which means forwards
//! it takes the older neighbour and backwards it takes the newer one, and a read
//! whose window opens on such a bucket answers not a number because nothing in
//! range came before it.
//!
//! # `EMPTY` fills gaps and only gaps
//!
//! A run of buckets with nothing in them is written out only when the whole
//! series has a real reading on both sides of it, which it looks for through the
//! two filters but not through the range. So a window sitting entirely inside a
//! gap is filled end to end, a run before the first reading the series ever held
//! is dropped, and a run after the last one is dropped as well.
//!
//! # The filter grammar
//!
//! Every command that searches on labels takes the same word and takes it apart
//! the same way, and the way is not a grammar either. It is four steps. An
//! argument holding `!=` anywhere is a negative one and splits on the two bytes
//! `!` and `=`, one holding `=` but not `!=` is a positive one and splits on
//! `=`, and one holding neither is `failed parsing labels`. Then, if there is a
//! `(` that is not the first byte and the byte before it is `=`, the argument is
//! the list form: it has to end in `)`, the label is the first field of the text
//! before the bracket split on the separator bytes, and the text between the
//! first `(` and the last `)` splits on commas with an empty field an error and
//! an empty whole an empty list. Otherwise the whole argument splits on the
//! separator bytes, the first field is the label and the second is the value,
//! and no second field means present for a negative and absent for a positive.
//!
//! Splitting is `strtok`, so a run of separators counts as one and everything
//! past the second field is dropped: `g==1`, `g===1` and `g=1=2` all ask the
//! same question as `g=1`, and `g!!=1` is a negative on `g`. The bracket test
//! reads the raw argument rather than a field, so `=(1)` is a list form with no
//! label and is an error while `()=1` is an ordinary positive on the label `()`,
//! and `g =(1)` asks about a label whose name ends in a space because nothing is
//! trimmed anywhere. A bracket that is not straight behind a separator is an
//! ordinary byte in an ordinary value, so `g=a(b` and `g=x()` are plain. Label
//! names and values are both compared byte for byte, so case matters on both.
//!
//! A search needs at least one predicate that says which series to take, and
//! only equals and the list form count, an empty list included. Anything else
//! is `please provide at least one matcher`. `TS.QUERYLABELS` is the exception
//! because its filter is optional, and with no filter at all it takes every
//! series and never asks that question.
//!
//! # The label surface walks the keyspace
//!
//! There is no index from a label to the series wearing it. The three commands
//! here walk every key in the keyspace, keep the ones that are series and ask
//! each one whether the rules hold, which is why they carry `bounded = "risk"`
//! in the registry. The reason is the one already written down in
//! `yo_kv::foreign`: a second table means `DEL`, `TYPE`, `EXISTS`, `KEYS`,
//! `SCAN`, `RANDOMKEY`, `EXPIRE`, `DBSIZE` and `FLUSHDB` each have to know about
//! it and stay in step with it, and nine places that have to agree will not. An
//! index can come later behind a measurement that says it is needed.
//!
//! # A label name written down twice
//!
//! A series is allowed to hold the same label name more than once and the two
//! ways of reading one back disagree about which value that is. `TS.MGET` with
//! `SELECTED_LABELS` gives the first in the order the labels were written.
//! `TS.QUERYLABELS VALUES` gives the smallest by byte order, so a series holding
//! `r=bb` and then `r=b` answers `b` on one and `bb` on the other. `TS.MGET`
//! with `WITHLABELS` sidesteps the question and writes every pair. Matching
//! looks at all of them, so such a series is found by either value.
//!
//! Two of the sentences the module raises about the keyword `SELECTED_LABELS`
//! spell it `SELECT_LABELS`, without the `ED`. That is copied as it stands.
//!
//! # The multi key range reads
//!
//! `TS.MRANGE` and `TS.MREVRANGE` are the single key read with a filter list in
//! front of them and an optional group behind them, so everything about buckets,
//! alignment, `COUNT` and the two sample filters is settled one series at a time
//! before a group ever sees a row. The only new arithmetic is the fold.
//!
//! The group grammar is three lookups rather than a parse. The `GROUPBY` is the
//! first one written and it has to sit behind the `FILTER` or the command is
//! refused for that alone. Its words are only a group when it is exactly four
//! from the end of the command, and until that is known its words are read as
//! filters, so `FILTER nope GROUPBY x REDUCE avg extra` answers about `nope`
//! rather than about the length. The reducer is the word behind the first
//! `REDUCE` anywhere in the command and not the word three past the `GROUPBY`,
//! which is worth knowing because a group on a label spelled `REDUCE` then
//! reduces by whatever word follows that one and usually fails.
//!
//! Ten reductions are allowed as a reducer, which is the fifteen a bucket takes
//! less the weighted mean, the first, the last and the two counts only a
//! compaction rule uses. A read already answering more than one number a row has
//! nothing to hand a reducer and is refused rather than reduced on one column.
//!
//! A group folds the rows its members answered by timestamp, and a timestamp
//! only one member answered for is still a row. Readings that are not numbers
//! are dropped before the reduction runs, so a moment where every member held an
//! empty bucket answers not a number, except a count, which answers zero. A
//! `COUNT` is applied to each member and then again to the fold, so a group can
//! answer fewer rows than the count asked for and never more. Series that do not
//! wear the label at all drop out of the reply entirely, a series wearing it
//! twice lands in the group its first value names, groups come out sorted by
//! name and the members of one come out sorted by key.
//!
//! The two protocols disagree about where the reducer and the member keys go.
//! RESP2 has nowhere to put them, so a `WITHLABELS` group writes them as two
//! more labels named `__reducer__` and `__source__`, the second holding the
//! member keys joined with commas. RESP3 writes them as two fields of their own,
//! `reducers` and `sources`, and leaves the labels holding only the pair the
//! group was made on. A `SELECTED_LABELS` on a group is looked up against that
//! one pair, so any other name comes back against a nil.
//!
//! # The joined reads
//!
//! `TS.NRANGE` and `TS.NREVRANGE` arrived in 8.10 and are the single key read
//! again, this time over a list of keys written behind a count, with the rows
//! lined up on the timestamp. A row is the timestamp and then one nested array
//! holding a reading for every column, in the order the keys were named, and a
//! key with no reading at a timestamp another key does have one at contributes a
//! not a number there. That nesting is the only shape in the family that is not
//! the flat pair, and it holds even when only one key was named. The same key
//! twice is allowed and answers twice.
//!
//! Because the keys sit behind a count the two are `movablekeys` and `COMMAND
//! GETKEYS` reads the count to find them, which is the same branch `MSETEX`
//! uses.
//!
//! An `AGGREGATION` here names one reduction for every key and then the one
//! bucket width, and each of those names may be a comma list, so a row can be
//! wider than the key count. Everything behind the width, meaning `EMPTY` and
//! `BUCKETTIMESTAMP`, is measured from the width slot rather than from the
//! keyword, so the single key placement rules simply shift along as keys are
//! added.
//!
//! Two orderings had to be measured rather than assumed. The reduction names are
//! read before the two ends of the span, unlike every other option, so a command
//! with both a bad timestamp and a bad reduction name answers about the name.
//! And `COUNT` applies to the joined rows rather than to each key, so every
//! series is read forward and in full, the join is built, and only then is it
//! turned around and cut.
//!
//! With exactly one key none of the reduction name checking applies at all and
//! the plain single key parser runs instead, which is why `AGGREGATION 100`
//! reads as a missing width there and as a count mismatch as soon as there are
//! two keys.
//!
//! `TS.READ` came with them and is the smallest read in the family: a key, one
//! timestamp, and every sample from there to the end. A dash means from the
//! beginning and a plus means the last sample only. It takes no options, has no
//! cap on how many rows come back, and is the one command here that answers the
//! server's own bare `WRONGTYPE` instead of the module's prefixed sentence.
//!
//! # Compaction rules
//!
//! `TS.CREATERULE` points a source at a destination and every reading the source
//! is given after that lands in a bucket of the destination. The rule has a
//! width, a reduction and an alignment, and the destination is an ordinary
//! series that a client can read, delete from and write to by hand, which is why
//! the destination's own duplicate policy decides what happens when the rule
//! writes over a bucket it already wrote.
//!
//! A rule never goes back over what the source held before it was made. It
//! carries the bucket it is filling and the first reading it was given in that
//! bucket, and it writes a bucket down when a reading arrives in a later one, so
//! the buckets the source already had are never written and the destination
//! starts from the next reading. A reading that arrives out of order into a
//! bucket the rule has already closed makes that bucket alone worked out again
//! from what the source now holds, and one that arrives into the bucket still
//! open makes the whole of that bucket count rather than only the readings that
//! came after the rule started it. `TS.DEL` on the source walks whatever the
//! destination holds over the span that was deleted, works each of those buckets
//! out again, drops the ones that ended up with nothing in them and drops
//! everything at or past the bucket the rule has open, because a delete can
//! reopen a bucket that had already been written.
//!
//! `LATEST` on any of the four reads shows the bucket a rule is still filling,
//! worked out from the source, as one more reading on the end. It is added
//! before any aggregation the read asked for, so a read that buckets its own
//! answer buckets that reading with the rest, and it does nothing at all on a
//! series no rule writes to. On the two multi key reads the word only counts in
//! front of the `FILTER`, and a `LATEST` inside a `SELECTED_LABELS` list is both
//! the flag and a label name that comes back against a nil.
//!
//! Two things a client can see differ from the module. A rebuilt bucket is not
//! carried into the bucket the rule has open, which is D-56, and a rename does
//! not follow the link, which is D-55.
//!
//! # Where the two protocols disagree
//!
//! A sample value is a simple string of the shortest digits that read back as
//! the same number on RESP2 and a RESP3 double on RESP3, which are two
//! renderings of one number: `1E-1` against `0.1`. `TS.INFO` is a flat array of
//! twenty eight on RESP2 and a map of fourteen on RESP3, and its labels follow
//! the same split one level down. The value half of the ignore window inside
//! `TS.INFO` is a plain double rather than the shortest digits, so half a degree
//! reads `0.5` there and `5E-1` out of `TS.GET`. That is two reply helpers
//! inside one module rather than a decision, and both are copied.
//!
//! # One thing the reference does not decide
//!
//! `TS.INCRBY key n TIMESTAMP` with nothing behind the keyword reads one past
//! the end of its own argument list on a real server, so what it answers depends
//! on what was in that memory: a fresh key gets `invalid timestamp`, a key
//! holding samples has been seen to get the backwards error and to get the
//! increment itself used as the timestamp. There is no behaviour there to copy,
//! so this answers `invalid timestamp` every time, which is what the reference
//! answers in the one case where it is reading memory it owns.
//!
//! # What a client can see that is different
//!
//! `TS.INFO` reports a memory usage of its own, which is D-53. It has to: the
//! number there is the module's own allocator arithmetic over a chunk layout
//! that is not this one, and an empty series here does not hold the four
//! kibibytes an empty series holds there.
//!
//! A read that would build more rows than yo will build is refused with one
//! sentence rather than attempted, which is D-54. The reference will happily try
//! to put a hundred million empty buckets in a reply and fall over somewhere
//! inside that, and asking for it is always a mistake, so this says so instead.
//!
//! One more thing shows up in a comparison and is not a difference. The eight
//! variance and standard deviation reductions are written the way the module
//! writes them, and a C compiler on arm64 contracts the last multiply and add of
//! that expression into a fused multiply add while one on x86-64 does not, so
//! the module answers two slightly different numbers on the two machines. What
//! this answers is the x86-64 one, on every platform.
//!
//! Everything else matches a real Redis 8.10.1 with RedisTimeSeries in it,
//! sample for sample and error for error.

use yo_common::num::{DOUBLE_MAX, parse_f64, parse_i64, write_dragonbox};
use yo_common::{Code, Error, Result};
use yo_kv::{Db, Foreign, KeyCursor, Keyspace, Kind};
use yo_series::{
    Agg, Buckets, Encoding, Policy, Query, Refused, Rows, Sample, Series, Stamp, Unread,
    bucket_start,
};
// The filter matchers below are called rules too, and both names come from the
// module rather than from here, so the one that travels furthest is renamed.
use yo_series::Rule as Compaction;

use super::args::{self, Args};
use super::table::Spec;
use crate::reply::Out;

/// What `TS.CREATE` says about a key that is already there, whatever it holds.
/// The existence is what is checked and not the type, so a key holding a string
/// gets this rather than `WRONGTYPE`.
const EXISTS: &[u8] = b"TSDB: key already exists";
/// What the commands that will not create one say about a key that is not
/// there.
const MISSING: &[u8] = b"TSDB: the key does not exist";
/// What the two that would have created one say about a key holding something
/// else. Every other command in the family answers `WRONGTYPE` for the same
/// key.
const NOT_A_SERIES: &[u8] = b"TSDB: the key is not a TSDB key";
/// A label with an empty name, an empty value, or a value holding one of the
/// three characters a filter expression is written with.
const BAD_LABELS: &[u8] = b"TSDB: Couldn't parse LABELS";
/// A retention that is missing or is not a whole number.
const BAD_RETENTION: &[u8] = b"TSDB: Couldn't parse RETENTION";
/// A chunk size that is missing or is not a whole number.
const BAD_CHUNK: &[u8] = b"TSDB: Couldn't parse CHUNK_SIZE";
/// One that is a number and is not a size a chunk may be.
const CHUNK_RANGE: &[u8] =
    b"TSDB: CHUNK_SIZE value must be a multiple of 8 in the range [48 .. 1048576]";
/// An encoding that is neither of the two.
const BAD_ENCODING: &[u8] = b"TSDB: unknown ENCODING parameter";
/// A duplicate policy keyword with nothing behind it.
const BAD_POLICY: &[u8] = b"TSDB: Couldn't parse DUPLICATE_POLICY";
/// One with a word behind it that is not a policy.
const UNKNOWN_POLICY: &[u8] = b"TSDB: Unknown DUPLICATE_POLICY";
/// An ignore window missing a half, or with a half that is not a number.
const BAD_IGNORE: &[u8] = b"TSDB: Couldn't parse IGNORE";
/// One whose halves are numbers and are below zero.
const NEGATIVE_IGNORE: &[u8] = b"TSDB: IGNORE arguments cannot be negative";
/// A timestamp that is not a whole number.
const BAD_TIMESTAMP: &[u8] = b"TSDB: invalid timestamp";
/// One that is a whole number and is below zero.
const NEGATIVE_TIMESTAMP: &[u8] = b"TSDB: invalid timestamp, must be a nonnegative integer";
/// A value that is not one the module's own reader accepts.
const BAD_VALUE: &[u8] = b"TSDB: invalid value";
/// An increment that is not a number.
const BAD_INCREMENT: &[u8] = b"TSDB: invalid increase/decrease value";
/// An increment onto a series whose newest value is not a number, which has no
/// answer to give.
const NAN_INCREMENT: &[u8] = b"TSDB: cannot increment/decrement NaN value";
/// A sample so far behind the newest one that retention has already gone past
/// where it would have landed.
const TOO_OLD: &[u8] = b"TSDB: Timestamp is older than retention";
/// A sample on a timestamp that is taken, under a policy that will not have it
/// replaced. One sentence for two rather different cases, which is the
/// module's.
const UPSERT: &[u8] = b"TSDB: Error at upsert, update is not supported when DUPLICATE_POLICY is set to BLOCK mode, or either current or new value is NaN and DUPLICATE_POLICY is MAX/MIN/SUM";
/// A `TS.DEL` whose first timestamp is not one.
const BAD_FROM: &[u8] = b"TSDB: wrong fromTimestamp";
/// One whose second is not.
const BAD_TO: &[u8] = b"TSDB: wrong toTimestamp";
/// A `COUNT` on the end of the command with nothing behind it.
const COUNT_MISSING: &[u8] = b"TSDB: COUNT argument is missing";
/// One with a word behind it that is not a whole number.
const BAD_COUNT: &[u8] = b"TSDB: Couldn't parse COUNT";
/// One that is a whole number and is below one.
const COUNT_RANGE: &[u8] = b"TSDB: Invalid COUNT value";
/// An `AGGREGATION` missing either half of the pair behind it, or one whose
/// bucket width is not a whole number.
const BAD_AGGREGATION: &[u8] = b"TSDB: Couldn't parse AGGREGATION";
/// A reduction list with nothing between two commas, or with nothing in it at
/// all.
const EMPTY_AGG: &[u8] = b"TSDB: Empty aggregation type in list";
/// One naming more reductions than a row will hold.
const TOO_MANY_AGGS: &[u8] = b"TSDB: Too many aggregation types";
/// One naming something that is not a reduction.
const UNKNOWN_AGG: &[u8] = b"TSDB: Unknown aggregation type";
/// A bucket width that is a whole number and is not above zero.
const BAD_BUCKET: &[u8] = b"TSDB: bucketDuration must be greater than zero";
/// An `EMPTY` anywhere other than the two places it is allowed to be.
const EMPTY_PLACE: &[u8] = b"TSDB: EMPTY flag should be the 3rd or 5th flag after AGGREGATION flag";
/// A `BUCKETTIMESTAMP` in the same position.
const BUCKET_TS_PLACE: &[u8] =
    b"TSDB: BUCKETTIMESTAMP flag should be the 3rd or 4th flag after AGGREGATION flag";
/// One with a word behind it that is not an end of a bucket.
const BAD_BUCKET_TS: &[u8] = b"TSDB: unknown BUCKETTIMESTAMP parameter";
/// An `ALIGN` with a word behind it that is neither end of the span nor a
/// timestamp.
const BAD_ALIGN: &[u8] = b"TSDB: unknown ALIGN parameter";
/// One on a read that is not cut into buckets, which has nothing to line up.
const ALIGN_NO_AGG: &[u8] = b"TSDB: ALIGN parameter can only be used with AGGREGATION";
/// `ALIGN start` on a read whose start is as far back as the series goes.
const ALIGN_START: &[u8] = b"TSDB: start alignment can only be used with explicit start timestamp";
/// `ALIGN end` on one whose end is as far forward as it goes.
const ALIGN_END: &[u8] = b"TSDB: end alignment can only be used with explicit end timestamp";
/// A `FILTER_BY_VALUE` without both of its ends behind it.
const FILTER_VALUE_MISSING: &[u8] = b"TSDB: FILTER_BY_VALUE one or more arguments are missing";
/// One whose lower end is not a number.
const BAD_MIN: &[u8] = b"TSDB: Couldn't parse MIN";
/// One whose upper end is not.
const BAD_MAX: &[u8] = b"TSDB: Couldn't parse MAX";
/// A `FILTER_BY_TS` with no timestamp behind it.
const FILTER_TS_MISSING: &[u8] = b"TSDB: FILTER_BY_TS one or more arguments are missing";
/// A read that would have built more rows than yo will build, which is D-54 and
/// is the one sentence here the module has no counterpart for.
const TOO_WIDE: &[u8] = b"TSDB: the requested range holds too many empty buckets";
/// A filter word that is not a predicate at all.
const BAD_FILTER: &[u8] = b"TSDB: failed parsing labels";
/// A filter list with nothing in it that says which series to take.
const NO_MATCHER: &[u8] = b"TSDB: please provide at least one matcher";
/// A `FILTER` on `TS.QUERYLABELS` with nothing behind it.
const NO_EXPRESSIONS: &[u8] = b"TSDB: FILTER given with no filter expressions";
/// A `TS.QUERYLABELS` whose first word is neither of the two it takes.
const BAD_SUBTYPE: &[u8] = b"TSDB: unknown subtype, must be one of LABELS|VALUES";
/// A word where `TS.QUERYLABELS` expects the one keyword it takes.
const EXPECTED_FILTER: &[u8] = b"TSDB: unknown argument, expected FILTER";
/// Both ways of asking for labels back at once. The module drops the `ED` from
/// one of the two keywords in its own sentence and this copies that.
const BOTH_LABELS: &[u8] = b"TSDB: cannot accept WITHLABELS and SELECT_LABELS together";
/// A `SELECTED_LABELS` with no label named behind it, spelled the same way.
const NO_SELECTED: &[u8] = b"TSDB: SELECT_LABELS should have at least 1 parameter";

/// A rule written on to a series that is already the destination of one.
const RULE_EXISTS: &[u8] = b"TSDB: the destination key already has a src rule";

/// A rule written out of a series that is itself a destination, which is the
/// module's way of saying a rule cannot be chained on to another.
const SOURCE_IS_DEST: &[u8] = b"TSDB: the source key already has a source rule";

/// A rule written into a series that is the source of one, which is the same
/// refusal seen from the other end.
const DEST_IS_SOURCE: &[u8] = b"TSDB: the destination key already has a dst rule";

/// A rule from a key to itself.
const SAME_KEY: &[u8] = b"TSDB: the source key and destination key should be different";

/// A rule taken off a pair that does not have one.
const NO_RULE: &[u8] = b"TSDB: compaction rule does not exist";

/// An alignment on a rule that is missing or is not a whole number at or above
/// zero.
const BAD_ALIGN_STAMP: &[u8] = b"TSDB: Couldn't parse alignTimestamp";

/// A third word on `TS.GET` that is not `LATEST`.
const THIRD_WORD: &[u8] = b"TSDB: wrong 3rd argument";

/// A joined read whose key count is missing, is not a whole number or is not
/// above zero.
const BAD_NUMKEYS: &[u8] = b"TSDB: numkeys must be a positive integer";

/// An `AGGREGATION` on a joined read that does not name one reduction for every
/// key it was given.
const AGG_NUMKEYS: &[u8] = b"TSDB: the number of AGGREGATION arguments must be equal to numkeys";

/// A `TS.READ` whose one timestamp is neither end of the series nor a timestamp
/// at or above zero. It goes out with nothing in front of it, the way the two
/// bare sentences above do.
const BAD_READ_AT: &[u8] = b"TSDB: invalid timestamp";

/// A multi key range read with no `FILTER` anywhere in it.
const MISSING_FILTER: &[u8] = b"TSDB: missing FILTER argument";

/// A `FILTER` on a multi key range read with nothing behind it. `TS.MGET` says
/// something else in the same situation and both sentences are copied.
const NO_FILTER_LABELS: &[u8] = b"TSDB: missing labels for filter argument";

/// A `GROUPBY` written in front of the filters rather than behind them.
const GROUPBY_ORDER: &[u8] = b"TSDB: GROUPBY should always come after filter";

/// A reducer name that is not one of the ten a group takes.
const BAD_REDUCER: &[u8] = b"TSDB: Invalid reducer type";

/// A group over a read that already answers more than one number a row.
const GROUPBY_COLUMNS: &[u8] =
    b"TSDB: GROUPBY is not allowed when multiple aggregators are specified";
/// A retention below zero, which the module writes with no prefix at all where
/// a retention that is not a number gets one.
const BARE_RETENTION: &[u8] = b"TSDB: Couldn't parse RETENTION";
/// A `TS.INCRBY` or `TS.DECRBY` whose timestamp is behind the newest sample,
/// which is the other one the module writes bare.
const BARE_BACKWARDS: &[u8] =
    b"TSDB: timestamp must be equal to or higher than the maximum existing timestamp";

/// What a key holding anything else gets from the seven commands that will not
/// create a series.
///
/// The word is inside the sentence rather than in front of it, so a client sees
/// `ERR WRONGTYPE ...` and not the bare `WRONGTYPE ...` every other command in
/// the server answers. That is the module writing its own error text and Redis
/// putting its own prefix on anything a module writes, and it is what a real
/// server sends, so it is what this sends.
const WRONG_KIND: &str = "WRONGTYPE Operation against a key holding the wrong kind of value";

/// The smallest a chunk may be.
const CHUNK_MIN: i64 = 48;
/// The largest.
const CHUNK_MAX: i64 = 1_048_576;

/// How many reductions one read may ask for.
const MAX_AGGS: usize = 16;
/// How many timestamps `FILTER_BY_TS` will read. The ones past this are left
/// where they are rather than being an error, which is the module's.
const MAX_FILTER_TS: usize = 128;
/// Where the two ends of the span sit on a single key read.
///
/// The option words start one slot past that. The module scans for them from
/// the `to` slot onwards, and that slot has already had to be a timestamp by the
/// time the scan runs, so nothing is ever found there. It is still where the
/// scan starts, so it is where this one starts too. A multi key read has no key
/// in front of the span and passes 1 instead.
const SPAN_AT: usize = 2;

/// A series under a key.
#[derive(Debug)]
pub(super) struct TsBody {
    /// The samples and the settings. Everything `TS.INFO` reports comes off it.
    s: Series,
}

impl Foreign for TsBody {
    fn type_name(&self) -> &'static str {
        // The module's own name for the type, hyphen and capitals included,
        // which is what a client sees from `TYPE` on a real server.
        "TSDB-TYPE"
    }

    fn encoding(&self) -> &'static str {
        "raw"
    }

    fn memory_bytes(&self) -> usize {
        self.s.memory_bytes()
    }

    fn is_empty(&self) -> bool {
        // A series with no samples in it is still a key. A client is expected to
        // create one and then write to it, and it would be a surprise to find it
        // gone in between.
        false
    }
}

/// Every command in the family, given the whole database.
///
/// Nothing below reaches a key on its own. The three helpers at the bottom of
/// the file are the only way into the keyspace, and each of them finds the
/// stripe its key is on and reads that one, so a command naming several series
/// touches a stripe per series and a rule linking two keys works whether or not
/// they landed on the same one.
pub(super) fn execute(db: &Db, spec: &Spec, args: Args<'_>, out: &mut Out) -> Result<()> {
    match spec.name {
        "ts.create" => create(db, &args, out),
        "ts.alter" => alter(db, &args, out),
        "ts.add" => add(db, &args, out),
        "ts.madd" => madd(db, &args, out),
        "ts.incrby" => incr(db, &args, out, true),
        "ts.decrby" => incr(db, &args, out, false),
        "ts.del" => del(db, &args, out),
        "ts.get" => get(db, &args, out),
        "ts.range" => range(db, &args, out, false),
        "ts.revrange" => range(db, &args, out, true),
        "ts.nrange" => nrange(db, &args, out, false),
        "ts.nrevrange" => nrange(db, &args, out, true),
        "ts.read" => tail(db, &args, out),
        "ts.queryindex" => queryindex(db, &args, out),
        "ts.querylabels" => querylabels(db, &args, out),
        "ts.mget" => mget(db, &args, out),
        "ts.createrule" => createrule(db, &args, out),
        "ts.deleterule" => deleterule(db, &args, out),
        "ts.mrange" => mrange(db, &args, out, false),
        "ts.mrevrange" => mrange(db, &args, out, true),
        "ts.info" => info(db, &args, out),
        other => unreachable!("{other} is not a time series command"),
    }
}

/// `TS.CREATE key [RETENTION n] [ENCODING e] [CHUNK_SIZE n] [DUPLICATE_POLICY p]
/// [IGNORE t v] [LABELS name value ...]`.
fn create(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let opts = match options(args) {
        Ok(opts) => opts,
        Err(bad) => return said(bad, "ts.create", out),
    };
    let key = args.get(1);
    if db.hold(key).kind_of(key).is_some() {
        return say(out, EXISTS);
    }
    let mut s = Series::new();
    apply(&mut s, opts);
    db.hold(key).put_foreign(key, Box::new(TsBody { s }));
    out.ok();
    Ok(())
}

/// `TS.ALTER key [RETENTION n] [CHUNK_SIZE n] [DUPLICATE_POLICY p] [IGNORE t v]
/// [LABELS name value ...]`.
///
/// An encoding is read and then thrown away, because a series that already holds
/// samples cannot be told to store them a different way and the module does not
/// try. Everything that was not named is left as it was.
fn alter(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let mut opts = match options(args) {
        Ok(opts) => opts,
        Err(bad) => return said(bad, "ts.alter", out),
    };
    opts.encoding = None;
    let key = args.get(1);
    let mut stripe = db.hold(key);
    let Some(body) = write(&mut stripe, key)? else {
        return say(out, MISSING);
    };
    apply(&mut body.s, opts);
    out.ok();
    Ok(())
}

/// `TS.ADD key timestamp value [option ...]`, which makes the series if it is
/// not there and reads the same options `TS.CREATE` does when it does.
///
/// On a series that is already there the only option that means anything is
/// `ON_DUPLICATE`, and the rest are read past. That is why a `DUPLICATE_POLICY`
/// on a `TS.ADD` against a key that exists does nothing at all.
fn add(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let key = args.get(1);
    let Some(value) = number(args.get(3)) else {
        return say(out, BAD_VALUE);
    };
    let at = match moment(db, args.get(2)) {
        Ok(at) => at,
        Err(msg) => return say(out, msg),
    };
    let over = if db.hold(key).kind_of(key).is_none() {
        let opts = match options(args) {
            Ok(opts) => opts,
            Err(bad) => return said(bad, "ts.add", out),
        };
        let mut s = Series::new();
        apply(&mut s, opts);
        db.hold(key).put_foreign(key, Box::new(TsBody { s }));
        None
    } else {
        if write(&mut db.hold(key), key).is_err() {
            return say(out, NOT_A_SERIES);
        }
        match find(args, b"ON_DUPLICATE") {
            None => None,
            Some(at) => match policy_at(args, at) {
                Ok(policy) => Some(policy),
                Err(bad) => return said(bad, "ts.add", out),
            },
        }
    };
    // The stripe goes back before the rules run, because a rule writes into
    // another key and that key can be on this same stripe.
    let stored = {
        let mut stripe = db.hold(key);
        let body = write(&mut stripe, key)?.expect("the series is there by now");
        let before = body.s.last();
        store(body, at, value, over, out).then_some(before)
    };
    if let Some(before) = stored {
        feed(db, key, at, before)?;
    }
    Ok(())
}

/// `TS.MADD key timestamp value [key timestamp value ...]`.
///
/// Every triple is answered in its own slot and a bad one does not stop the ones
/// after it, so this is the only command in the family whose reply can hold both
/// timestamps and errors. It creates nothing: a key that is not already a series
/// is an error in its slot.
fn madd(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    if args.len() < 4 || !(args.len() - 1).is_multiple_of(3) {
        return Err(args::wrong_arity("ts.madd"));
    }
    out.array((args.len() - 1) / 3);
    for i in (1..args.len()).step_by(3) {
        let Some(value) = number(args.get(i + 2)) else {
            out.error_line(b"ERR ", BAD_VALUE);
            continue;
        };
        let at = match moment(db, args.get(i + 1)) {
            Ok(at) => at,
            Err(msg) => {
                out.error_line(b"ERR ", msg);
                continue;
            }
        };
        let key = args.get(i);
        let stored = {
            let mut stripe = db.hold(key);
            match write(&mut stripe, key) {
                Ok(Some(body)) => {
                    let before = body.s.last();
                    store(body, at, value, None, out).then_some(before)
                }
                Ok(None) | Err(_) => {
                    out.error_line(b"ERR ", NOT_A_SERIES);
                    None
                }
            }
        };
        if let Some(before) = stored {
            feed(db, key, at, before)?;
        }
    }
    Ok(())
}

/// `TS.INCRBY key n [TIMESTAMP t] [option ...]` and `TS.DECRBY`, which are one
/// command with the sign flipped.
///
/// This is an add whose value is the newest value plus or minus the number
/// given, so it only ever writes at or past the newest sample and a timestamp
/// behind that is an error rather than a backfill. The sample goes in under the
/// last policy whatever the series says, which is what makes two of these on one
/// timestamp add up rather than collide.
fn incr(db: &Db, args: &Args<'_>, out: &mut Out, up: bool) -> Result<()> {
    let key = args.get(1);
    let name = if up { "ts.incrby" } else { "ts.decrby" };
    let exists = db.hold(key).kind_of(key).is_some();
    if exists {
        // A key holding something else is `WRONGTYPE` here rather than the
        // sentence `TS.ADD` gives, and it is answered before the number is even
        // looked at.
        write(&mut db.hold(key), key)?;
    }
    let Some(by) = parse_f64(args.get(2)).filter(|n| !n.is_nan()) else {
        return say(out, BAD_INCREMENT);
    };

    // `LABELS` swallows the rest of the command, so a `TIMESTAMP` behind one is
    // a label name and not the keyword. Both are looked for past the key and the
    // number so that a key called `TIMESTAMP` stays a key.
    let labels_at = find_from(args, 3, b"LABELS");
    let stamp_at =
        find_from(args, 3, b"TIMESTAMP").filter(|&at| labels_at.is_none_or(|labels| at < labels));
    let at = match stamp_at {
        None => now(db),
        Some(at) => match args.opt(at + 1) {
            None => return say(out, BAD_TIMESTAMP),
            Some(b"*") => now(db),
            Some(word) => match parse_i64(word) {
                Some(at) => at,
                None => return say(out, BAD_TIMESTAMP),
            },
        },
    };

    if !exists {
        let opts = match options(args) {
            Ok(opts) => opts,
            Err(bad) => return said(bad, name, out),
        };
        let mut s = Series::new();
        apply(&mut s, opts);
        db.hold(key).put_foreign(key, Box::new(TsBody { s }));
    }
    // As `TS.ADD`, the stripe goes back before the rules run.
    let stored = {
        let mut stripe = db.hold(key);
        let body = write(&mut stripe, key)?.expect("the series is there by now");
        let last = body.s.last_sample();
        if last.is_some_and(|s| at < s.at) {
            out.error(BARE_BACKWARDS);
            return Ok(());
        }
        let base = last.map_or(0.0, |s| s.value);
        if base.is_nan() {
            return say(out, NAN_INCREMENT);
        }
        let value = if up { base + by } else { base - by };
        let before = body.s.last();
        store(body, at, value, Some(Policy::Last), out).then_some(before)
    };
    if let Some(before) = stored {
        feed(db, key, at, before)?;
    }
    Ok(())
}

/// `TS.DEL key from to`, both ends included, which answers how many samples
/// went.
fn del(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let Some(from) = span(args.get(2), b"-", 0) else {
        return say(out, BAD_FROM);
    };
    let Some(to) = span(args.get(3), b"+", i64::MAX) else {
        return say(out, BAD_TO);
    };
    let key = args.get(1);
    // The count comes out of the block and the stripe goes back with it,
    // because putting the rules back in step writes into other keys.
    let gone = {
        let mut stripe = db.hold(key);
        let Some(body) = write(&mut stripe, key)? else {
            return say(out, MISSING);
        };
        body.s.delete(from, to)
    };
    // The rules are put back in step even when the delete took nothing, because
    // the reference starts the open bucket again either way and a fold that had
    // been counting part of a bucket goes back to counting all of it.
    undo(db, key, from, to)?;
    out.uint(gone as u64);
    Ok(())
}

/// `TS.GET key [LATEST]`, which is the newest sample, or an empty array when
/// there is not one.
///
/// The three checks run in an order of their own: a fourth word is an arity
/// error before the key is looked at, the key is resolved before the third word
/// is read, and only then is a third word that is not `LATEST` complained about.
fn get(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    if args.len() > 3 {
        return Err(args::wrong_arity("ts.get"));
    }
    let latest = args.opt(2).is_some_and(|word| args::is(word, b"LATEST"));
    let open = if latest {
        open_bucket(db, args.get(1))?
    } else {
        None
    };
    let key = args.get(1);
    let mut stripe = db.hold(key);
    let Some(body) = read(&mut stripe, key)? else {
        return say(out, MISSING);
    };
    if args.len() == 3 && !latest {
        return say(out, THIRD_WORD);
    }
    match open.or_else(|| body.s.last_sample()) {
        None => out.array(0),
        Some(sample) => {
            out.array(2);
            out.int(sample.at);
            value(out, sample.value);
        }
    }
    Ok(())
}

/// `TS.RANGE key from to [LATEST] [FILTER_BY_TS ts ...] [FILTER_BY_VALUE min
/// max] [COUNT n] [[ALIGN a] AGGREGATION spec width [BUCKETTIMESTAMP b]
/// [EMPTY]]` and `TS.REVRANGE`, which are one read in two directions.
///
/// The key is resolved before a single option word is looked at, so a read of a
/// key that is not there says so whatever else is wrong with the command. After
/// that the two ends of the span are read, and only then the options, in the
/// order their errors come out in.
fn range(db: &Db, args: &Args<'_>, out: &mut Out, reverse: bool) -> Result<()> {
    let name = if reverse { "ts.revrange" } else { "ts.range" };
    let key = args.get(1);
    if read(&mut db.hold(key), key)?.is_none() {
        return say(out, MISSING);
    }
    let mut query = match reading(args, reverse, SPAN_AT) {
        Ok(query) => query,
        Err(bad) => return said(bad, name, out),
    };
    if find_from(args, SPAN_AT + 1, b"LATEST").is_some() {
        query.latest = open_bucket(db, key)?;
    }
    let mut stripe = db.hold(key);
    let body = read(&mut stripe, key)?.expect("the series was there a moment ago");
    let rows = match body.s.read(&query) {
        Ok(rows) => rows,
        Err(Unread::TooWide) => return say(out, TOO_WIDE),
    };
    spread(out, &rows);
    Ok(())
}

/// `TS.NRANGE numkeys key [key ...] from to [LATEST] [FILTER_BY_TS ts ...]
/// [FILTER_BY_VALUE min max] [COUNT n] [[ALIGN a] AGGREGATION spec ... width
/// [BUCKETTIMESTAMP b] [EMPTY]]` and `TS.NREVRANGE`, which read the same span
/// out of several named series and line the answers up on their timestamps.
///
/// Every timestamp any of the series answered gets one row, with each series
/// contributing the numbers it was asked for and a NaN for each of them when it
/// answered nothing at that timestamp. `COUNT` is taken off the joined rows and
/// not off each series, so a count of one is one row and not one row a key.
///
/// The checks run in an order of their own again: the key count, then the arity
/// the key count implies, then the reduction names, then the two ends of the
/// span, then the rest of the options, then the keys. Only the names come out in
/// front of the span, which is why they are read here and then read again below
/// with everything else.
fn nrange(db: &Db, args: &Args<'_>, out: &mut Out, reverse: bool) -> Result<()> {
    let name = if reverse { "ts.nrevrange" } else { "ts.nrange" };
    let Some(keys) = parse_i64(args.get(1))
        .filter(|&n| n > 0)
        .and_then(|n| usize::try_from(n).ok())
    else {
        return say(out, BAD_NUMKEYS);
    };
    // The span sits behind the keys, and a command with no room for both ends of
    // it is an arity error however many keys it named.
    let span = 2 + keys;
    if span + 1 >= args.len() {
        return Err(args::wrong_arity(name));
    }
    if keys > 1
        && let Some(at) = find_from(args, span + 1, b"AGGREGATION")
        && let Err(bad) = agg_lists(args, at, keys)
    {
        return said(bad, name, out);
    }
    let (query, lists) = match reading_keys(args, reverse, span, keys) {
        Ok(read) => read,
        Err(bad) => return said(bad, name, out),
    };
    for i in 0..keys {
        let key = args.get(2 + i);
        if read(&mut db.hold(key), key)?.is_none() {
            return say(out, MISSING);
        }
    }

    let latest = find_from(args, span + 1, b"LATEST").is_some();
    let mut taken = Vec::with_capacity(keys);
    for (i, list) in lists.iter().enumerate() {
        let key = args.get(2 + i).to_vec();
        let open = if latest { open_bucket(db, &key)? } else { None };
        // Each series is read oldest first and without the count, because the
        // rows are turned around and cut down after they have been lined up.
        let mut one = Query {
            reverse: false,
            count: None,
            latest: open,
            ..query.clone()
        };
        if let Some(buckets) = one.buckets.as_mut() {
            buckets.aggs.clone_from(list);
        }
        let mut stripe = db.hold(&key);
        let body = read(&mut stripe, &key)?.expect("the series was there a moment ago");
        match body.s.read(&one) {
            Ok(rows) => taken.push(rows),
            Err(Unread::TooWide) => return say(out, TOO_WIDE),
        }
    }

    let mut rows = join(&taken);
    if reverse {
        rows.flip();
    }
    if let Some(n) = query.count {
        rows.keep(n);
    }
    columns(out, &rows);
    Ok(())
}

/// Several reads lined up on their timestamps.
///
/// The reads are all oldest first here, so this is one cursor into each of them
/// and one pass: the oldest timestamp any cursor is looking at is the next row,
/// every read sitting on it hands over its numbers and steps forward, and every
/// read that is not fills its share of the row with NaN.
fn join(taken: &[Rows]) -> Rows {
    let width = taken.iter().map(|rows| rows.width).sum();
    let mut rows = Rows {
        width,
        ..Rows::default()
    };
    let mut at = vec![0usize; taken.len()];
    while let Some(now) = taken
        .iter()
        .zip(&at)
        .filter_map(|(one, &i)| one.stamps.get(i).copied())
        .min()
    {
        rows.stamps.push(now);
        for (k, one) in taken.iter().enumerate() {
            if one.stamps.get(at[k]) == Some(&now) {
                rows.values.extend_from_slice(one.row(at[k]));
                at[k] += 1;
            } else {
                rows.values
                    .extend(core::iter::repeat_n(f64::NAN, one.width));
            }
        }
    }
    rows
}

/// `TS.READ key from`, every sample from a timestamp to the end of the series.
///
/// The arity in the table is `-3` and the module answers a wrong arity to
/// anything longer than three words, so the second half of that is checked here.
/// Two of its answers are unlike the rest of the family: a key that is not there
/// answers an empty array rather than a sentence, and a key holding something
/// else answers the server's own bare `WRONGTYPE` rather than the module's
/// prefixed one, because this command lets the server report the type where the
/// others write their own sentence about it.
fn tail(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    if args.len() != 3 {
        return Err(args::wrong_arity("ts.read"));
    }
    let word = args.get(2);
    let newest = word == b"+";
    let from = if word == b"-" || newest {
        0
    } else {
        match parse_i64(word).filter(|&n| n >= 0) {
            Some(at) => at,
            None => return said(Bad::Bare(BAD_READ_AT), "ts.read", out),
        }
    };
    let key = args.get(1);
    let mut stripe = db.hold(key);
    let Some(body) = bare_read(&mut stripe, key)? else {
        out.array(0);
        return Ok(());
    };
    // The far end of the series is the last sample and not a timestamp, so it is
    // looked up rather than parsed, and a series holding nothing has no far end
    // to look up.
    let from = match (newest, body.s.last_sample()) {
        (true, None) => {
            out.array(0);
            return Ok(());
        }
        (true, Some(last)) => last.at,
        (false, _) => from,
    };
    let query = Query {
        from,
        to: i64::MAX,
        ..Query::default()
    };
    let rows = body
        .s
        .read(&query)
        .expect("a read with no buckets fills no gaps");
    spread(out, &rows);
    Ok(())
}

/// Everything a read asked for, gathered off the command.
fn reading(args: &Args<'_>, reverse: bool, at: usize) -> core::result::Result<Query, Bad> {
    reading_keys(args, reverse, at, 1).map(|(query, _)| query)
}

/// The same for a read over `keys` series, which names one reduction list a key
/// where a single key read names one.
fn reading_keys(
    args: &Args<'_>,
    reverse: bool,
    at: usize,
    keys: usize,
) -> core::result::Result<(Query, Vec<Vec<Agg>>), Bad> {
    let opts = at + 1;
    // Which end was written open matters later: an alignment against an end
    // that was never named is an error, and only the one character counts as
    // naming nothing.
    let open_start = args.get(at) == b"-";
    let Some(from) = span(args.get(at), b"-", 0) else {
        return Err(Bad::Said(BAD_FROM));
    };
    let open_end = args.get(at + 1) == b"+";
    let Some(to) = span(args.get(at + 1), b"+", i64::MAX) else {
        return Err(Bad::Said(BAD_TO));
    };

    // LATEST comes first in the module and means read through to the compaction
    // rule that feeds this series. Nothing has rules yet, so it is read for the
    // order of the errors around it and then dropped.
    let count = count_of(args, opts, keys)?;
    let read = buckets_of(args, opts, keys)?;
    let lists = read
        .as_ref()
        .map_or_else(|| vec![Vec::new(); keys], |(_, lists)| lists.clone());
    let mut buckets = read.map(|(buckets, _)| buckets);
    if let Some(align) = align_of(
        args,
        opts,
        buckets.is_some(),
        open_start,
        open_end,
        from,
        to,
    )? && let Some(buckets) = buckets.as_mut()
    {
        buckets.align = align;
    }
    Ok((
        Query {
            from,
            to,
            reverse,
            count,
            by_ts: by_ts(args, opts)?,
            by_value: by_value(args, opts)?,
            buckets,
            latest: None,
        },
        lists,
    ))
}

/// How many rows at most, if the read said.
///
/// A `COUNT` sitting where a reduction name or a reducer name goes is that name
/// and not the keyword, so the scan starts again past it. A joined read names
/// one reduction a key, so there are that many slots behind `AGGREGATION` to
/// step over rather than one, and a reducer is one more slot after that.
fn count_of(args: &Args<'_>, opts: usize, keys: usize) -> core::result::Result<Option<usize>, Bad> {
    let Some(mut at) = find_from(args, opts, b"COUNT") else {
        return Ok(None);
    };
    while find_from(args, opts, b"AGGREGATION").is_some_and(|agg| at > agg && at <= agg + keys)
        || find_from(args, opts, b"REDUCE") == Some(at - 1)
    {
        match find_from(args, at + 1, b"COUNT") {
            Some(next) => at = next,
            None => return Ok(None),
        }
    }
    if at + 1 == args.len() {
        return Err(Bad::Said(COUNT_MISSING));
    }
    let Some(n) = parse_i64(args.get(at + 1)) else {
        return Err(Bad::Said(BAD_COUNT));
    };
    if n < 1 {
        return Err(Bad::Said(COUNT_RANGE));
    }
    Ok(Some(usize::try_from(n).unwrap_or(usize::MAX)))
}

/// How a read is cut into buckets, and the reduction list every key it names was
/// given. A single key read has one list and a joined read has one a key.
type Bucketing = (Buckets, Vec<Vec<Agg>>);

/// How the read is cut into buckets, if it asked to be.
///
/// `EMPTY` and `BUCKETTIMESTAMP` are only looked for when there is an
/// `AGGREGATION` to hang them off, and each has to sit a fixed number of words
/// behind it. That is not a grammar, it is a pair of arithmetic checks against
/// where the keyword was found, and a command that puts the words in the
/// documented order passes both.
fn buckets_of(
    args: &Args<'_>,
    opts: usize,
    keys: usize,
) -> core::result::Result<Option<Bucketing>, Bad> {
    let Some(at) = find_from(args, opts, b"AGGREGATION") else {
        return Ok(None);
    };
    // One name a key and then the width, so the width moves along as the keys
    // are added and the two flags behind it move with it.
    let width = at + 1 + keys;
    if width >= args.len() {
        return Err(Bad::Said(BAD_AGGREGATION));
    }
    let Some(delta) = parse_i64(args.get(width)) else {
        return Err(Bad::Said(BAD_AGGREGATION));
    };
    // The names are read before the width is looked at, so a bad name on a zero
    // width bucket answers about the name.
    let lists = agg_lists(args, at, keys)?;
    if delta <= 0 {
        return Err(Bad::Said(BAD_BUCKET));
    }

    let mut buckets = Buckets {
        aggs: lists[0].clone(),
        delta,
        align: 0,
        empty: false,
        stamp: Stamp::Start,
    };
    if let Some(flag) = find_from(args, opts, b"EMPTY") {
        if flag != width + 1 && flag != width + 3 {
            return Err(Bad::Said(EMPTY_PLACE));
        }
        buckets.empty = true;
    }
    if let Some(flag) = find_from(args, opts, b"BUCKETTIMESTAMP") {
        if flag != width + 1 && flag != width + 2 {
            return Err(Bad::Said(BUCKET_TS_PLACE));
        }
        if flag + 1 >= args.len() {
            return Err(Bad::Arity);
        }
        let word = args.get(flag + 1);
        buckets.stamp = if args::is(word, b"start") || word == b"-" {
            Stamp::Start
        } else if args::is(word, b"end") || word == b"+" {
            Stamp::End
        } else if args::is(word, b"mid") || word == b"~" {
            Stamp::Mid
        } else {
            return Err(Bad::Said(BAD_BUCKET_TS));
        };
    }
    Ok(Some((buckets, lists)))
}

/// The reduction list every key of a joined read was given, which is one word a
/// key in front of the bucket width.
///
/// A single key read is the plain grammar and none of the counting below applies
/// to it: one word, whatever it holds, and then the width. Past one key the
/// module counts what it was given, and it counts by what the words look like. A
/// whole number where a name goes is the bucket width arriving early and means
/// the command named fewer reductions than it has keys, a name where the width
/// goes means it named more, and both of those are the one sentence about the
/// count. A word in a name slot that is neither a number nor a reduction is
/// about the name instead, which is why a name that is simply wrong answers
/// differently from one that is missing.
fn agg_lists(args: &Args<'_>, at: usize, keys: usize) -> core::result::Result<Vec<Vec<Agg>>, Bad> {
    if keys == 1 {
        return Ok(vec![reductions(args.get(at + 1))?]);
    }
    let mut lists = Vec::with_capacity(keys);
    for i in 0..keys {
        let Some(word) = args.opt(at + 1 + i) else {
            return Err(Bad::Said(AGG_NUMKEYS));
        };
        if parse_i64(word).is_some() {
            return Err(Bad::Said(AGG_NUMKEYS));
        }
        lists.push(reductions(word)?);
    }
    if args
        .opt(at + 1 + keys)
        .is_some_and(|word| reductions(word).is_ok())
    {
        return Err(Bad::Said(AGG_NUMKEYS));
    }
    Ok(lists)
}

/// The reduction list, which is one name or several separated by commas.
///
/// A name may appear twice and gets a column each time, because the list is
/// read as written rather than gathered into a set.
fn reductions(spec: &[u8]) -> core::result::Result<Vec<Agg>, Bad> {
    let mut aggs = Vec::new();
    for word in spec.split(|&b| b == b',') {
        if word.is_empty() {
            return Err(Bad::Said(EMPTY_AGG));
        }
        if aggs.len() >= MAX_AGGS {
            return Err(Bad::Said(TOO_MANY_AGGS));
        }
        let Some(agg) = Agg::parse(word) else {
            return Err(Bad::Said(UNKNOWN_AGG));
        };
        aggs.push(agg);
    }
    Ok(aggs)
}

/// The timestamp the bucket edges line up against, if the read named one.
///
/// The word is read before any of the three things that make an alignment
/// wrong are checked, so a word that is not an alignment at all answers about
/// the word rather than about the missing `AGGREGATION`.
fn align_of(
    args: &Args<'_>,
    opts: usize,
    bucketed: bool,
    open_start: bool,
    open_end: bool,
    from: i64,
    to: i64,
) -> core::result::Result<Option<i64>, Bad> {
    let Some(at) = find_from(args, opts, b"ALIGN") else {
        return Ok(None);
    };
    if at + 1 >= args.len() {
        return Err(Bad::Arity);
    }
    let word = args.get(at + 1);
    let start = args::is(word, b"start") || word == b"-";
    let end = args::is(word, b"end") || word == b"+";
    let align = if start {
        from
    } else if end {
        to
    } else {
        match parse_i64(word).filter(|&n| n >= 0) {
            Some(n) => n,
            None => return Err(Bad::Said(BAD_ALIGN)),
        }
    };
    if !bucketed {
        return Err(Bad::Said(ALIGN_NO_AGG));
    }
    if start && open_start {
        return Err(Bad::Said(ALIGN_START));
    }
    if end && open_end {
        return Err(Bad::Said(ALIGN_END));
    }
    Ok(Some(align))
}

/// The two ends of the value filter, if the read named them.
fn by_value(args: &Args<'_>, opts: usize) -> core::result::Result<Option<(f64, f64)>, Bad> {
    let Some(at) = find_from(args, opts, b"FILTER_BY_VALUE") else {
        return Ok(None);
    };
    if at + 2 >= args.len() {
        return Err(Bad::Said(FILTER_VALUE_MISSING));
    }
    let Some(min) = parse_f64(args.get(at + 1)) else {
        return Err(Bad::Said(BAD_MIN));
    };
    let Some(max) = parse_f64(args.get(at + 2)) else {
        return Err(Bad::Said(BAD_MAX));
    };
    Ok(Some((min, max)))
}

/// The timestamps the read will take, if it listed any.
///
/// The list runs until a word that is not a timestamp, which is how the option
/// after it is found, so a list that runs to the end of the command is a list
/// and a list followed by `COUNT` stops at the keyword.
fn by_ts(args: &Args<'_>, opts: usize) -> core::result::Result<Option<Vec<i64>>, Bad> {
    let Some(at) = find_from(args, opts, b"FILTER_BY_TS") else {
        return Ok(None);
    };
    if at + 1 == args.len() {
        return Err(Bad::Said(FILTER_TS_MISSING));
    }
    let mut list = Vec::new();
    let mut i = at + 1;
    while i < args.len() && list.len() < MAX_FILTER_TS {
        match parse_i64(args.get(i)).filter(|&n| n >= 0) {
            Some(n) => list.push(n),
            None => break,
        }
        i += 1;
    }
    if list.is_empty() {
        return Err(Bad::Said(FILTER_TS_MISSING));
    }
    list.sort_unstable();
    list.dedup();
    Ok(Some(list))
}

/// One predicate off one filter word.
///
/// Six of them, and the split that matters is not equal against not equal: it is
/// whether the predicate says which series to take or only which ones to leave
/// out. A filter list made only of the second kind is refused, because it would
/// otherwise mean the whole keyspace.
#[derive(Debug)]
enum Rule {
    /// `label=value`.
    Is(Vec<u8>, Vec<u8>),
    /// `label!=value`.
    IsNot(Vec<u8>, Vec<u8>),
    /// `label=(a,b,c)`, and an empty list is allowed and matches nothing.
    In(Vec<u8>, Vec<Vec<u8>>),
    /// `label!=(a,b,c)`.
    NotIn(Vec<u8>, Vec<Vec<u8>>),
    /// `label!=`, which is the series that have the label at all.
    Present(Vec<u8>),
    /// `label=`, which is the series that do not.
    Absent(Vec<u8>),
}

impl Rule {
    /// Whether this predicate says which series to take rather than which to
    /// leave out.
    fn positive(&self) -> bool {
        matches!(self, Rule::Is(..) | Rule::In(..))
    }

    /// Whether a series wearing `labels` passes.
    ///
    /// A label name can be written down twice on one series and both stay, so
    /// every one of these looks at all the values under the name rather than the
    /// first, and the two negatives are the positives turned round rather than
    /// their own walk.
    fn holds(&self, labels: &[(Vec<u8>, Vec<u8>)]) -> bool {
        let any = |name: &Vec<u8>, take: &dyn Fn(&Vec<u8>) -> bool| {
            labels.iter().any(|(n, v)| n == name && take(v))
        };
        match self {
            Rule::Is(name, want) => any(name, &|v| v == want),
            Rule::IsNot(name, want) => !any(name, &|v| v == want),
            Rule::In(name, list) => any(name, &|v| list.contains(v)),
            Rule::NotIn(name, list) => !any(name, &|v| list.contains(v)),
            Rule::Present(name) => any(name, &|_| true),
            Rule::Absent(name) => !any(name, &|_| true),
        }
    }
}

/// One filter word turned into a predicate.
///
/// The module reads these with `strtok`, which skips a run of separators rather
/// than seeing an empty field between two of them, so `room==kitchen` is
/// `room=kitchen` and `room=kitchen=x` is too. The list form is the one place
/// that is not true: there the label is whatever comes before the bracket, taken
/// apart the same way, and the value is the raw text between the brackets. A
/// bracket that is not straight after the separator is an ordinary character in
/// an ordinary value, which is why `room=a(b` is a filter for the value `a(b`
/// and `room=(a,b` is a malformed list.
fn rule(arg: &[u8]) -> core::result::Result<Rule, Bad> {
    let negated = arg.windows(2).any(|w| w == b"!=");
    let separator: &[u8] = if negated {
        b"!="
    } else if arg.contains(&b'=') {
        b"="
    } else {
        return Err(Bad::Said(BAD_FILTER));
    };
    if let Some(open) = arg.iter().position(|&b| b == b'(')
        && open > 0
        && arg[open - 1] == b'='
    {
        if arg.last() != Some(&b')') {
            return Err(Bad::Said(BAD_FILTER));
        }
        let mut at = 0;
        let Some(name) = word(&arg[..open], separator, &mut at) else {
            return Err(Bad::Said(BAD_FILTER));
        };
        let inside = &arg[open + 1..arg.len() - 1];
        let mut list = Vec::new();
        if !inside.is_empty() {
            for part in inside.split(|&b| b == b',') {
                if part.is_empty() {
                    return Err(Bad::Said(BAD_FILTER));
                }
                list.push(part.to_vec());
            }
        }
        let name = name.to_vec();
        return Ok(if negated {
            Rule::NotIn(name, list)
        } else {
            Rule::In(name, list)
        });
    }
    let mut at = 0;
    let Some(name) = word(arg, separator, &mut at) else {
        return Err(Bad::Said(BAD_FILTER));
    };
    let name = name.to_vec();
    match word(arg, separator, &mut at) {
        Some(value) => {
            let value = value.to_vec();
            Ok(if negated {
                Rule::IsNot(name, value)
            } else {
                Rule::Is(name, value)
            })
        }
        None if negated => Ok(Rule::Present(name)),
        None => Ok(Rule::Absent(name)),
    }
}

/// The next run of `arg` that is not made of separators, starting at `at` and
/// leaving `at` past it. This is `strtok` and nothing more.
fn word<'a>(arg: &'a [u8], separator: &[u8], at: &mut usize) -> Option<&'a [u8]> {
    while *at < arg.len() && separator.contains(&arg[*at]) {
        *at += 1;
    }
    let start = *at;
    while *at < arg.len() && !separator.contains(&arg[*at]) {
        *at += 1;
    }
    (start < *at).then(|| &arg[start..*at])
}

/// A whole filter list, which has to name at least one thing to take.
fn rules(args: &Args<'_>, from: usize) -> core::result::Result<Vec<Rule>, Bad> {
    rules_in(args, from, args.len())
}

/// The same over a span of the command rather than everything behind a word,
/// which is what a range read needs because its group sits past its filters.
fn rules_in(args: &Args<'_>, from: usize, to: usize) -> core::result::Result<Vec<Rule>, Bad> {
    let mut list = Vec::with_capacity(to.saturating_sub(from));
    for i in from..to {
        list.push(rule(args.get(i))?);
    }
    if !list.iter().any(Rule::positive) {
        return Err(Bad::Said(NO_MATCHER));
    }
    Ok(list)
}

/// Every series the filter list takes, by key name, in order.
///
/// This walks the keyspace rather than reading a label index, and that is on
/// purpose. An index beside the keyspace is a second table that `DEL`, `EXPIRE`,
/// `RENAME`, `COPY`, `RESTORE`, `FLUSHDB` and eviction all have to remember to
/// keep in step, and the one that forgets leaves a filter answering a key that
/// is not there any more. The walk cannot drift because there is nothing for it
/// to drift from, and it only ever touches the keys holding a foreign body,
/// which on a keyspace that is mostly strings and hashes is a small part of it.
///
/// The walk covers every stripe, because a filter takes the series it names
/// wherever they landed, and the names come back sorted so a database that is
/// one stripe wide and one that is many answer in the same order.
fn chosen(db: &Db, rules: &[Rule]) -> Vec<Vec<u8>> {
    let mut names = Vec::new();
    db.scan(KeyCursor::START, usize::MAX, Some(Kind::Foreign), |key| {
        names.push(key.to_vec());
    });
    names.retain(|name| {
        db.hold(name)
            .foreign(name)
            .ok()
            .flatten()
            .and_then(<dyn Foreign>::downcast_ref::<TsBody>)
            .is_some_and(|body| rules.iter().all(|r| r.holds(body.s.labels())))
    });
    names.sort_unstable();
    names
}

/// `TS.QUERYINDEX filter [filter ...]`, the key names a filter list takes.
fn queryindex(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let rules = match rules(args, 1) {
        Ok(rules) => rules,
        Err(bad) => return said(bad, "ts.queryindex", out),
    };
    let names = chosen(db, &rules);
    out.set(names.len());
    for name in &names {
        out.bulk(name);
    }
    Ok(())
}

/// `TS.QUERYLABELS LABELS [FILTER filter ...]` for the label names in use, and
/// `TS.QUERYLABELS VALUES label [FILTER filter ...]` for the values one of them
/// takes.
///
/// This is the one command in the family where a filter list is optional, and a
/// missing one means every series rather than none, so the rule about naming at
/// least one thing to take does not apply until a `FILTER` is written down.
fn querylabels(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let values = if args::is(args.get(1), b"LABELS") {
        false
    } else if args::is(args.get(1), b"VALUES") {
        true
    } else {
        return say(out, BAD_SUBTYPE);
    };
    // `VALUES` needs the label it is asking about, and the module counts that
    // itself rather than leaving it to the arity in the table.
    let at = if values { 3 } else { 2 };
    if args.len() < at {
        return Err(args::wrong_arity("ts.querylabels"));
    }
    let mut rules = Vec::new();
    if args.len() > at {
        if !args::is(args.get(at), b"FILTER") {
            return say(out, EXPECTED_FILTER);
        }
        if args.len() == at + 1 {
            return say(out, NO_EXPRESSIONS);
        }
        rules = match self::rules(args, at + 1) {
            Ok(rules) => rules,
            Err(bad) => return said(bad, "ts.querylabels", out),
        };
    }
    let wanted = if values {
        args.get(2).to_vec()
    } else {
        Vec::new()
    };
    let mut found: Vec<Vec<u8>> = Vec::new();
    for name in chosen(db, &rules) {
        let mut stripe = db.hold(&name);
        let Some(body) = read(&mut stripe, &name)? else {
            continue;
        };
        if values {
            // A series that writes the same label name down twice contributes
            // one value here and it is the smallest of them, which is not the
            // first the way `SELECTED_LABELS` is. The module reads this side
            // off a form sorted by name and then by value and stops at the
            // first row that matches, so a later duplicate never shows up.
            let smallest = body
                .s
                .labels()
                .iter()
                .filter(|(label, _)| *label == wanted)
                .map(|(_, value)| value)
                .min();
            if let Some(value) = smallest {
                found.push(value.clone());
            }
        } else {
            for (label, _) in body.s.labels() {
                found.push(label.clone());
            }
        }
    }
    found.sort_unstable();
    found.dedup();
    out.set(found.len());
    for word in &found {
        out.bulk(word);
    }
    Ok(())
}

/// Which labels a multi key read writes back beside each series.
#[derive(Default)]
enum Wearing {
    /// None, which is what a read that asked for nothing gets.
    #[default]
    None,
    /// All of them.
    All,
    /// These, in this order, with a nil where the series has no such label.
    These(Vec<Vec<u8>>),
}

/// `TS.MGET [LATEST] [WITHLABELS | SELECTED_LABELS label ...] FILTER filter ...`,
/// the newest sample of every series a filter list takes.
///
/// `FILTER` is not optional and its absence is an arity error rather than a
/// syntax one, however many words the command has, because the module counts the
/// arguments it needed after it has found the keyword rather than before.
fn mget(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let Some(filter) = find(args, b"FILTER") else {
        return Err(args::wrong_arity("ts.mget"));
    };
    let wearing = match wearing(args) {
        Ok(wearing) => wearing,
        Err(bad) => return said(bad, "ts.mget", out),
    };
    let rules = match rules(args, filter + 1) {
        Ok(rules) => rules,
        Err(bad) => return said(bad, "ts.mget", out),
    };
    let names = chosen(db, &rules);
    let latest = latest(args);
    if out.proto().is_resp3() {
        out.map(names.len());
    } else {
        out.array(names.len());
    }
    for name in &names {
        let open = if latest { open_bucket(db, name)? } else { None };
        let mut stripe = db.hold(name);
        let Some(body) = read(&mut stripe, name)? else {
            continue;
        };
        let last = open.or_else(|| body.s.last_sample());
        let labels = body.s.labels().to_vec();
        // RESP3 has the key as the map key and the rest as a pair behind it.
        // RESP2 has no map, so the key goes inside a triple with the other two.
        if out.proto().is_resp3() {
            out.bulk(name);
            out.array(2);
        } else {
            out.array(3);
            out.bulk(name);
        }
        wearing.write(out, &labels);
        match last {
            None => out.array(0),
            Some(sample) => {
                out.array(2);
                out.int(sample.at);
                value(out, sample.value);
            }
        }
    }
    Ok(())
}

/// The words that end a `SELECTED_LABELS` list.
///
/// Everything after the keyword is a label name until one of these turns up,
/// and `LATEST`, `EMPTY`, `BUCKETTIMESTAMP` and any word the command does not
/// know are not among them, so all four are read as label names and come back
/// against a nil. `TS.MGET` and the two multi key range reads share this list
/// even though half of it means nothing to `TS.MGET`.
const ENDS_LABELS: &[&[u8]] = &[
    b"FILTER",
    b"FILTER_BY_TS",
    b"FILTER_BY_VALUE",
    b"COUNT",
    b"AGGREGATION",
    b"ALIGN",
    b"WITHLABELS",
    b"REDUCE",
    b"GROUPBY",
];

/// Whether a multi key read asked for the bucket its sources are still filling.
///
/// The word only counts in front of the filters, because everything past those
/// is read as a filter. A `SELECTED_LABELS` list does not stop it: a `LATEST`
/// inside one is both the keyword and a label name, and comes back against a nil
/// as well as turning the flag on.
fn latest(args: &Args<'_>) -> bool {
    let stop = find(args, b"FILTER").unwrap_or_else(|| args.len());
    find(args, b"LATEST").is_some_and(|at| at < stop)
}

/// Which labels a multi key read asked for.
///
/// Both keywords are looked for across the whole command rather than in front of
/// `FILTER`, so a `WITHLABELS` written behind the filters still collides with a
/// `SELECTED_LABELS` written in front of them.
fn wearing(args: &Args<'_>) -> core::result::Result<Wearing, Bad> {
    let all = find(args, b"WITHLABELS").is_some();
    let some = find(args, b"SELECTED_LABELS");
    if all && some.is_some() {
        return Err(Bad::Said(BOTH_LABELS));
    }
    if let Some(at) = some {
        let mut list: Vec<Vec<u8>> = Vec::new();
        for i in at + 1..args.len() {
            if ENDS_LABELS.iter().any(|word| args::is(args.get(i), word)) {
                break;
            }
            list.push(args.get(i).to_vec());
        }
        if list.is_empty() {
            return Err(Bad::Said(NO_SELECTED));
        }
        return Ok(Wearing::These(list));
    }
    Ok(if all { Wearing::All } else { Wearing::None })
}

impl Wearing {
    /// The labels half of one series in a multi key reply.
    ///
    /// A map on RESP3 and a list of pairs on RESP2, which is the same split
    /// `TS.INFO` makes one level down and not the flat array a map downgrades
    /// to on its own.
    fn write(&self, out: &mut Out, labels: &[(Vec<u8>, Vec<u8>)]) {
        let pairs: Vec<(&[u8], Option<&[u8]>)> = match self {
            Wearing::None => Vec::new(),
            Wearing::All => labels
                .iter()
                .map(|(n, v)| (n.as_slice(), Some(v.as_slice())))
                .collect(),
            Wearing::These(wanted) => wanted
                .iter()
                .map(|name| {
                    let found = labels.iter().find(|(n, _)| n == name);
                    (name.as_slice(), found.map(|(_, v)| v.as_slice()))
                })
                .collect(),
        };
        let resp3 = out.proto().is_resp3();
        if resp3 {
            out.map(pairs.len());
        } else {
            out.array(pairs.len());
        }
        for (name, value) in pairs {
            if !resp3 {
                out.array(2);
            }
            out.bulk(name);
            match value {
                Some(value) => out.bulk(value),
                None => out.nil(),
            }
        }
    }
}

/// One series a multi key read took, which is its key, the labels it wears and
/// the rows it answered.
type Took = (Vec<u8>, Vec<(Vec<u8>, Vec<u8>)>, Rows);

/// One group a multi key read made, which is the value its members share, the
/// keys of those members and a set of rows each.
type Group = (Vec<u8>, Vec<Vec<u8>>, Vec<Rows>);

/// The ten reductions a group is allowed to ask for, which is the fifteen a
/// bucket takes less the weighted mean, the first, the last and the two counts
/// that only the compaction rules use.
const REDUCERS: &[Agg] = &[
    Agg::Avg,
    Agg::Sum,
    Agg::Min,
    Agg::Max,
    Agg::Range,
    Agg::Count,
    Agg::StdP,
    Agg::StdS,
    Agg::VarP,
    Agg::VarS,
];

/// `TS.MRANGE from to [FILTER_BY_TS t...] [FILTER_BY_VALUE min max] [WITHLABELS
/// | SELECTED_LABELS l...] [COUNT n] [[ALIGN a] AGGREGATION spec width
/// [BUCKETTIMESTAMP b] [EMPTY]] FILTER f... [GROUPBY label REDUCE reducer]` and
/// `TS.MREVRANGE`, which read a span out of every series a filter list takes.
///
/// The span itself is the same read `TS.RANGE` does, one series at a time, so
/// everything about buckets, alignment and the two sample filters is already
/// settled by the time this starts. What is new is the filter list in front of
/// it and the group behind it.
fn mrange(db: &Db, args: &Args<'_>, out: &mut Out, reverse: bool) -> Result<()> {
    let name = if reverse { "ts.mrevrange" } else { "ts.mrange" };
    // The order the checks happen in, which is the order their errors come out
    // in and is not the order the words are written in.
    let query = match reading(args, reverse, SPAN_AT - 1) {
        Ok(query) => query,
        Err(bad) => return said(bad, name, out),
    };
    let Some(filter) = find(args, b"FILTER") else {
        return say(out, MISSING_FILTER);
    };
    let wearing = match wearing(args) {
        Ok(wearing) => wearing,
        Err(bad) => return said(bad, name, out),
    };
    let group = find(args, b"GROUPBY");
    if group.is_some_and(|at| at < filter) {
        return say(out, GROUPBY_ORDER);
    }
    // The filters stop where the group starts, and a `GROUPBY` written anywhere
    // other than four words from the end is not a group at all yet, so its words
    // are read as filters and answer about themselves first.
    let end = group.unwrap_or_else(|| args.len());
    if filter + 1 == end {
        return say(out, NO_FILTER_LABELS);
    }
    let rules = match rules_in(args, filter + 1, end) {
        Ok(rules) => rules,
        Err(bad) => return said(bad, name, out),
    };
    // Only now is a group made to be the last four words of the command, so a
    // second one or a word past the reducer is an arity error rather than
    // anything that names what went wrong.
    if group.is_some_and(|at| at + 4 != args.len()) {
        return Err(args::wrong_arity(name));
    }
    // The reducer is the word behind the first `REDUCE` anywhere in the command
    // rather than the word three past the `GROUPBY`, so a group whose label is
    // spelled `REDUCE` reduces by whatever word follows that one.
    let reducer = match group {
        None => None,
        Some(_) => {
            let word = find(args, b"REDUCE")
                .filter(|at| at + 1 < args.len())
                .map_or(&[][..], |at| args.get(at + 1));
            match Agg::parse(word).filter(|agg| REDUCERS.contains(agg)) {
                Some(agg) => Some(agg),
                None => return say(out, BAD_REDUCER),
            }
        }
    };
    // A read answering more than one number a row has nothing sensible to hand
    // a reducer, and the module says so rather than picking a column.
    if reducer.is_some() && query.buckets.as_ref().is_some_and(|b| b.aggs.len() > 1) {
        return say(out, GROUPBY_COLUMNS);
    }

    let names = chosen(db, &rules);
    let latest = latest(args);
    let mut query = query;
    let mut taken: Vec<Took> = Vec::with_capacity(names.len());
    for key in names {
        query.latest = if latest { open_bucket(db, &key)? } else { None };
        let mut stripe = db.hold(&key);
        let Some(body) = read(&mut stripe, &key)? else {
            continue;
        };
        let labels = body.s.labels().to_vec();
        let rows = match body.s.read(&query) {
            Ok(rows) => rows,
            Err(Unread::TooWide) => return say(out, TOO_WIDE),
        };
        taken.push((key, labels, rows));
    }
    match (group, reducer) {
        (Some(at), Some(agg)) => {
            grouped(out, args.get(at + 1), agg, &wearing, &query, taken);
        }
        _ => plainly(out, &wearing, &query, &taken),
    }
    Ok(())
}

/// One series a row, which is what a read with no group answers.
fn plainly(out: &mut Out, wearing: &Wearing, query: &Query, taken: &[Took]) {
    let resp3 = out.proto().is_resp3();
    if resp3 {
        out.map(taken.len());
    } else {
        out.array(taken.len());
    }
    for (key, labels, rows) in taken {
        // RESP3 puts the key in front as the map key and answers three things
        // behind it, one of which RESP2 has no room for and does not write.
        if resp3 {
            out.bulk(key);
            out.array(3);
            wearing.write(out, labels);
            named(out, b"aggregators", query);
        } else {
            out.array(3);
            out.bulk(key);
            wearing.write(out, labels);
        }
        spread(out, rows);
    }
}

/// The reductions a read asked for, which only RESP3 writes back and which is an
/// empty list on a read that asked for no bucketing at all.
fn named(out: &mut Out, under: &[u8], query: &Query) {
    let aggs = query.buckets.as_ref().map_or(&[][..], |b| &b.aggs);
    out.map(1);
    out.bulk(under);
    out.array(aggs.len());
    for agg in aggs {
        out.bulk(agg.name().as_bytes());
    }
}

/// One row a group, each folded out of every series wearing the same value of
/// one label.
///
/// A series that does not wear the label at all is not in any group and drops
/// out of the reply. A series that wears it twice is in the group its first
/// value names, the same rule `SELECTED_LABELS` follows.
fn grouped(
    out: &mut Out,
    label: &[u8],
    agg: Agg,
    wearing: &Wearing,
    query: &Query,
    taken: Vec<Took>,
) {
    let mut groups: Vec<Group> = Vec::new();
    for (key, labels, rows) in taken {
        let Some((_, value)) = labels.iter().find(|(n, _)| n == label) else {
            continue;
        };
        let at = match groups.iter().position(|(v, _, _)| v == value) {
            Some(at) => at,
            None => {
                groups.push((value.clone(), Vec::new(), Vec::new()));
                groups.len() - 1
            }
        };
        groups[at].1.push(key);
        groups[at].2.push(rows);
    }
    groups.sort_by(|a, b| a.0.cmp(&b.0));

    let resp3 = out.proto().is_resp3();
    if resp3 {
        out.map(groups.len());
    } else {
        out.array(groups.len());
    }
    for (value, sources, rows) in &groups {
        let mut title = label.to_vec();
        title.push(b'=');
        title.extend_from_slice(value);
        // The one label a group wears is the one it was grouped on. RESP2 has
        // nowhere else to put the reducer and the sources, so it writes them as
        // two more labels, and RESP3 writes them as two fields of their own.
        let mut labels = vec![(label.to_vec(), value.clone())];
        if !resp3 && matches!(wearing, Wearing::All) {
            labels.push((b"__reducer__".to_vec(), agg.name().as_bytes().to_vec()));
            labels.push((b"__source__".to_vec(), sources.join(&b","[..])));
        }
        if resp3 {
            out.bulk(&title);
            out.array(4);
            wearing.write(out, &labels);
            out.map(1);
            out.bulk(b"reducers");
            out.array(1);
            out.bulk(agg.name().as_bytes());
            out.map(1);
            out.bulk(b"sources");
            out.array(sources.len());
            for key in sources {
                out.bulk(key);
            }
        } else {
            out.array(3);
            out.bulk(&title);
            wearing.write(out, &labels);
        }
        spread(out, &folded(rows, agg, query));
    }
}

/// Every series in one group folded into one row a timestamp.
///
/// A timestamp only one series answered for is still a row, reduced over the one
/// reading it holds, and a `COUNT` is applied again over the fold because the
/// read that built each set already applied it to that set.
fn folded(taken: &[Rows], agg: Agg, query: &Query) -> Rows {
    let mut all: Vec<(i64, f64)> = Vec::new();
    for rows in taken {
        for i in 0..rows.len() {
            all.push((rows.stamps[i], rows.row(i)[0]));
        }
    }
    all.sort_by_key(|&(at, _)| at);
    let mut stamps = Vec::new();
    let mut values = Vec::new();
    let mut at = 0;
    while at < all.len() {
        let moment = all[at].0;
        let mut past = at;
        while past < all.len() && all[past].0 == moment {
            past += 1;
        }
        let readings: Vec<f64> = all[at..past].iter().map(|&(_, v)| v).collect();
        stamps.push(moment);
        values.push(yo_series::group(agg, &readings));
        at = past;
    }
    if query.reverse {
        stamps.reverse();
        values.reverse();
    }
    if let Some(n) = query.count {
        stamps.truncate(n);
        values.truncate(n);
    }
    Rows {
        stamps,
        values,
        width: 1,
    }
}

/// The rows half of a reply, a timestamp and its numbers a row.
fn spread(out: &mut Out, rows: &Rows) {
    out.array(rows.len());
    for i in 0..rows.len() {
        let row = rows.row(i);
        out.array(1 + row.len());
        out.int(rows.stamps[i]);
        for &d in row {
            value(out, d);
        }
    }
}

/// The same rows the way a joined read writes them, which is the timestamp and
/// then one nested array holding a reading for every column.
///
/// A single sample read writes the timestamp and the one value side by side, so
/// the two shapes are not the same even when a joined read was given one key.
fn columns(out: &mut Out, rows: &Rows) {
    out.array(rows.len());
    for i in 0..rows.len() {
        let row = rows.row(i);
        out.array(2);
        out.int(rows.stamps[i]);
        out.array(row.len());
        for &d in row {
            value(out, d);
        }
    }
}

/// `TS.CREATERULE source dest AGGREGATION reduction bucket [align]`, which sets
/// a series to be folded into another one as it is written to.
///
/// Both ends of the link are stored, the source keeping a rule for each of its
/// destinations and each destination keeping the name of its source, because
/// `TS.INFO` reports both sides and neither can work the other out on its own.
///
/// The checks run in an order that has nothing to do with the order the words
/// are written in. The bucket width is read before the reduction name, the
/// reduction name before the width is compared against zero, and all of that
/// before either key is looked up.
fn createrule(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    if !matches!(args.len(), 6 | 7) || !args::is(args.get(3), b"AGGREGATION") {
        return Err(args::wrong_arity("ts.createrule"));
    }
    let Some(delta) = parse_i64(args.get(5)) else {
        return say(out, BAD_AGGREGATION);
    };
    let Some(agg) = Agg::parse(args.get(4)) else {
        return say(out, UNKNOWN_AGG);
    };
    if delta <= 0 {
        return say(out, BAD_BUCKET);
    }
    let align = match args.opt(6) {
        None => 0,
        Some(word) => match parse_i64(word).filter(|&n| n >= 0) {
            Some(n) => n,
            None => return say(out, BAD_ALIGN_STAMP),
        },
    };

    let source = args.get(1).to_vec();
    let dest = args.get(2).to_vec();
    if source == dest {
        return say(out, SAME_KEY);
    }
    // Both keys are pruned before either is read, so a rule whose other end has
    // been deleted is out of the way by the time this asks whether there is one.
    prune(db, &source)?;
    prune(db, &dest)?;
    // The two ends are two keys and can be one stripe, so each end is looked at
    // inside a block of its own and the stripe goes back before the other end is
    // reached for. The order the checks come out in is what it was.
    {
        let mut stripe = db.hold(&source);
        let Some(body) = read(&mut stripe, &source)? else {
            return say(out, MISSING);
        };
        if body.s.source().is_some() {
            return say(out, SOURCE_IS_DEST);
        }
    }
    {
        let mut stripe = db.hold(&dest);
        let Some(body) = write(&mut stripe, &dest)? else {
            return say(out, MISSING);
        };
        if !body.s.rules().is_empty() {
            return say(out, DEST_IS_SOURCE);
        }
        if body.s.source().is_some() {
            return say(out, RULE_EXISTS);
        }
        body.s.set_source(Some(source.clone()));
    }
    let mut stripe = db.hold(&source);
    let body = write(&mut stripe, &source)?.expect("the source was there a moment ago");
    // A new rule starts with no bucket open, which is what keeps it off the
    // samples the source already held. Those buckets are never written and the
    // destination only starts filling from the next sample the source is given.
    body.s.add_rule(Compaction {
        dest,
        delta,
        agg,
        align,
        open: None,
        start: 0,
    });
    out.ok();
    Ok(())
}

/// `TS.DELETERULE source dest`, which takes the link apart from both ends.
fn deleterule(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    if args.len() != 3 {
        return Err(args::wrong_arity("ts.deleterule"));
    }
    let source = args.get(1).to_vec();
    let dest = args.get(2).to_vec();
    prune(db, &source)?;
    {
        let mut stripe = db.hold(&source);
        let Some(body) = write(&mut stripe, &source)? else {
            return say(out, MISSING);
        };
        if !body.s.drop_rule(&dest) {
            return say(out, NO_RULE);
        }
    }
    let mut stripe = db.hold(&dest);
    if let Some(body) = write(&mut stripe, &dest)? {
        body.s.set_source(None);
    }
    out.ok();
    Ok(())
}

/// Forgets the rules whose other end is gone.
///
/// Nothing tells this module that a key has been deleted, so a link that has
/// lost one of its ends is found rather than reported. Every command that reads
/// a link or writes one asks for this first, which is enough for a client to
/// never see a rule that has stopped meaning anything.
fn prune(db: &Db, key: &[u8]) -> Result<()> {
    // A key holding something else has no rules to lose and is not this
    // function's business to complain about, since the command that asked has
    // its own order to complain in.
    // What the rules say is copied out and the stripe goes back, because
    // reading the other end of a rule can want this same stripe again.
    let (source, dests) = {
        let mut stripe = db.hold(key);
        let Ok(Some(body)) = read(&mut stripe, key) else {
            return Ok(());
        };
        let source = body.s.source().map(<[u8]>::to_vec);
        let dests: Vec<Vec<u8>> = body
            .s
            .rules()
            .iter()
            .map(|rule| rule.dest.clone())
            .collect();
        (source, dests)
    };
    let mut kept = Vec::with_capacity(dests.len());
    for dest in dests {
        if points_at(db, &dest, key)? {
            kept.push(dest);
        }
    }
    let orphan = match &source {
        None => false,
        Some(source) => !feeds(db, source, key)?,
    };
    let mut stripe = db.hold(key);
    let Some(body) = write(&mut stripe, key)? else {
        return Ok(());
    };
    body.s.keep_rules(&kept);
    if orphan {
        body.s.set_source(None);
    }
    Ok(())
}

/// Whether the series under `key` is a destination fed by `source`.
fn points_at(db: &Db, key: &[u8], source: &[u8]) -> Result<bool> {
    let mut stripe = db.hold(key);
    let body = match read(&mut stripe, key) {
        Ok(body) => body,
        // A key that has been replaced by something else is not a destination
        // any more, and saying so here is what takes the rule off the source.
        Err(_) => return Ok(false),
    };
    Ok(body.is_some_and(|body| body.s.source() == Some(source)))
}

/// Whether the series under `key` holds a rule writing into `dest`.
fn feeds(db: &Db, key: &[u8], dest: &[u8]) -> Result<bool> {
    let mut stripe = db.hold(key);
    let body = match read(&mut stripe, key) {
        Ok(body) => body,
        Err(_) => return Ok(false),
    };
    Ok(body.is_some_and(|body| body.s.rules().iter().any(|rule| rule.dest == dest)))
}

/// One folded bucket read off a source series, or `None` when nothing counted
/// towards it.
///
/// The fold starts at `from` rather than at the bucket edge, because a rule only
/// counts the samples it has been given and a sample written into a bucket
/// before the rule existed is not one of them. With `from` on the edge this is
/// the plain bucketed read `TS.RANGE` does, which is what makes a bucket worked
/// out again agree with a read of the source down to the last digit.
fn fold(s: &Series, rule: &Compaction, at: i64, from: i64) -> Option<f64> {
    // The weighted mean weighs its last reading against whichever comes first,
    // the end of the bucket or the end of the read, and a rule always gets the
    // end of the bucket, so it is given a bucket of room on the far side and the
    // first row is the answer. Nothing else looks past the bucket it is in.
    // It also measures the bucket from its own edge rather than from the first
    // reading a rule was given in it, so the whole bucket goes into the answer
    // whatever the rule has seen of it.
    let wide = rule.agg == Agg::Twa;
    let room = if wide { 2 } else { 1 };
    let query = Query {
        from: if wide { at } else { from },
        to: at.saturating_add(rule.delta * room) - 1,
        buckets: Some(Buckets {
            aggs: vec![rule.agg],
            delta: rule.delta,
            align: rule.align,
            empty: false,
            stamp: Stamp::Start,
        }),
        ..Query::default()
    };
    let rows = s.read(&query).ok()?;
    // A bucket edge below the epoch is reported as sitting on it, and the room
    // given to the weighted mean can put a second bucket in the answer, so the
    // row is picked by its timestamp rather than by its place.
    let want = at.max(0);
    (0..rows.len())
        .find(|&i| rows.stamps[i] == want)
        .map(|i| rows.row(i)[0])
}

/// Moves every rule on `key` along after one sample landed on it.
///
/// A rule holds one bucket open and writes it down when a sample past it
/// arrives, which is why a destination trails its source by a bucket and why the
/// samples the source held before the rule was made never turn up in it. A
/// sample landing behind the newest one is the other half of this: the bucket it
/// falls in is closed already, so it is worked out again from everything the
/// source holds there and written over whatever the destination had.
fn feed(db: &Db, key: &[u8], at: i64, before: Option<i64>) -> Result<()> {
    for rule in rules_of(db, key)? {
        // What the source says is worked out first and the stripe goes back
        // with the answer, because the destination can be on this same stripe.
        let (written, moved) = {
            let mut stripe = db.hold(key);
            let Some(body) = read(&mut stripe, key)? else {
                return Ok(());
            };
            let landed = bucket_start(at, rule.delta, rule.align);
            let mut written = None;
            let mut moved = None;
            if before.is_none_or(|last| at >= last) {
                match rule.open {
                    // The bucket that was open is finished, so it is written
                    // down and the new one takes its place.
                    Some(open) if landed > open => {
                        written = fold(&body.s, &rule, open, rule.start).map(|value| (open, value));
                        moved = Some((Some(landed), at));
                    }
                    Some(_) => {}
                    None => moved = Some((Some(landed), at)),
                }
            } else {
                let newest = bucket_start(before.unwrap_or(at), rule.delta, rule.align);
                if landed < newest {
                    written = fold(&body.s, &rule, landed, landed).map(|value| (landed, value));
                } else if rule.open == Some(landed) {
                    // The open bucket counts this one too, and the module works
                    // the whole bucket out again rather than folding one more
                    // reading into what it had, so from here on the fold counts
                    // the lot.
                    moved = Some((Some(landed), landed));
                }
            }
            (written, moved)
        };
        let mut stripe = db.hold(&rule.dest);
        if let Some((at, value)) = written
            && let Some(dest) = write(&mut stripe, &rule.dest)?
        {
            // An alignment can put the first bucket edge before zero, and the
            // module writes that bucket at zero rather than at a timestamp no
            // series is allowed to hold.
            let at = at.max(0);
            // The destination's own policy has no say here: a bucket that has
            // been worked out again replaces the one written before it whatever
            // that policy is.
            let _ = dest.s.add(Sample::new(at, value), Some(Policy::Last));
        }
        drop(stripe);
        let mut stripe = db.hold(key);
        if let Some((open, start)) = moved
            && let Some(body) = write(&mut stripe, key)?
            && let Some(mine) = body.s.rule_mut(&rule.dest)
        {
            mine.open = open;
            mine.start = start;
        }
    }
    Ok(())
}

/// Puts every destination of `key` back in step after samples were taken out of
/// it between `from` and `to`.
///
/// Deleting works the other way round from adding. Nothing new can appear in a
/// destination, so this walks what the destination already holds over the span
/// rather than the span itself, which is what keeps a delete of everything from
/// costing a bucket per millisecond. A bucket the delete emptied goes, one it
/// only thinned is worked out again, and one the delete reopened by taking the
/// newest samples out from under it goes as well. The open bucket is thrown away
/// and started again over everything the source has left there, so a fold that
/// had been counting only the samples the rule was given goes back to counting
/// the lot.
fn undo(db: &Db, key: &[u8], from: i64, to: i64) -> Result<()> {
    for rule in rules_of(db, key)? {
        // Source and destination are two keys and can be one stripe, so each
        // step takes the one it needs and gives it back before the next.
        let open = {
            let mut stripe = db.hold(key);
            let Some(body) = read(&mut stripe, key)? else {
                return Ok(());
            };
            body.s
                .last()
                .map(|last| bucket_start(last, rule.delta, rule.align))
        };
        let head = bucket_start(from.max(0), rule.delta, rule.align);
        let tail = to.saturating_add(rule.delta);
        let stamps: Vec<i64> = {
            let mut stripe = db.hold(&rule.dest);
            let Some(dest) = read(&mut stripe, &rule.dest)? else {
                continue;
            };
            dest.s.range(head, tail).map(|sample| sample.at).collect()
        };
        let rows = {
            let mut stripe = db.hold(key);
            let body = read(&mut stripe, key)?.expect("the source was there a moment ago");
            let mut rows = Vec::with_capacity(stamps.len());
            for at in stamps {
                let value = match open {
                    Some(open) if at < open => fold(&body.s, &rule, at, at),
                    _ => None,
                };
                rows.push((at, value));
            }
            rows
        };
        {
            let mut stripe = db.hold(&rule.dest);
            for (at, value) in rows {
                if let Some(dest) = write(&mut stripe, &rule.dest)? {
                    match value {
                        Some(value) => {
                            let _ = dest.s.add(Sample::new(at, value), Some(Policy::Last));
                        }
                        None => {
                            dest.s.delete(at, at);
                        }
                    }
                }
            }
            if let Some(dest) = write(&mut stripe, &rule.dest)? {
                // Whatever the span was, a destination never holds the bucket
                // its source is still filling, and a delete can make an older
                // bucket that one.
                match open {
                    Some(open) => dest.s.delete(open, i64::MAX),
                    None => dest.s.delete(0, i64::MAX),
                };
            }
        }
        let mut stripe = db.hold(key);
        if let Some(body) = write(&mut stripe, key)?
            && let Some(mine) = body.s.rule_mut(&rule.dest)
        {
            mine.open = open;
            mine.start = open.unwrap_or(0);
        }
    }
    Ok(())
}

/// The rules on `key`, with the ones whose destination is gone already dropped.
fn rules_of(db: &Db, key: &[u8]) -> Result<Vec<Compaction>> {
    prune(db, key)?;
    Ok(read(&mut db.hold(key), key)?.map_or_else(Vec::new, |body| body.s.rules().to_vec()))
}

/// The bucket a destination's source is still filling, which is what `LATEST`
/// asks to see.
///
/// It is only there on a series that is the destination of a rule whose source
/// has given it a sample since the rule was made. Everywhere else `LATEST` means
/// nothing at all and the read answers what it would have answered without it.
fn open_bucket(db: &Db, key: &[u8]) -> Result<Option<Sample>> {
    prune(db, key)?;
    let source = {
        let mut stripe = db.hold(key);
        let Some(body) = read(&mut stripe, key)? else {
            return Ok(None);
        };
        let Some(source) = body.s.source().map(<[u8]>::to_vec) else {
            return Ok(None);
        };
        source
    };
    let mut stripe = db.hold(&source);
    let Some(body) = read(&mut stripe, &source)? else {
        return Ok(None);
    };
    let Some(rule) = body.s.rules().iter().find(|rule| rule.dest == key).cloned() else {
        return Ok(None);
    };
    let Some(open) = rule.open else {
        return Ok(None);
    };
    Ok(fold(&body.s, &rule, open, rule.start).map(|value| Sample::new(open.max(0), value)))
}

/// `TS.INFO key`, which is fourteen fields about the series and takes no field
/// name.
fn info(db: &Db, args: &Args<'_>, out: &mut Out) -> Result<()> {
    let resp3 = out.proto().is_resp3();
    let key = args.get(1);
    prune(db, key)?;
    let mut stripe = db.hold(key);
    let Some(body) = read(&mut stripe, key)? else {
        return say(out, MISSING);
    };
    let s = &body.s;
    let (ignore_time, ignore_value) = s.ignore();
    out.map(14);
    out.simple(b"totalSamples");
    out.uint(s.len() as u64);
    out.simple(b"memoryUsage");
    out.uint(s.memory_bytes() as u64);
    // A series with nothing in it reports zero for both ends rather than saying
    // it has no ends, and so does one everything has been deleted from.
    out.simple(b"firstTimestamp");
    out.int(s.first().unwrap_or(0));
    out.simple(b"lastTimestamp");
    out.int(s.last().unwrap_or(0));
    out.simple(b"retentionTime");
    out.int(s.retention());
    out.simple(b"chunkCount");
    out.uint(s.chunk_count() as u64);
    out.simple(b"chunkSize");
    out.uint(s.chunk_bytes() as u64);
    out.simple(b"chunkType");
    out.simple(s.encoding().name().as_bytes());
    // Never a nil: a series that has not been told what to do about a repeated
    // timestamp reports the default rather than nothing.
    out.simple(b"duplicatePolicy");
    out.simple(s.policy().unwrap_or(Policy::Block).name().as_bytes());
    out.simple(b"labels");
    if resp3 {
        out.map(s.labels().len());
        for (name, value) in s.labels() {
            out.bulk(name);
            out.bulk(value);
        }
    } else {
        out.array(s.labels().len());
        for (name, value) in s.labels() {
            out.array(2);
            out.bulk(name);
            out.bulk(value);
        }
    }
    out.simple(b"sourceKey");
    match s.source() {
        Some(key) => out.bulk(key),
        None => out.nil(),
    }
    // RESP2 writes a rule as four things with the destination first and RESP3
    // makes the destination the key of a map, which is the same split the labels
    // above go through one level up.
    out.simple(b"rules");
    if resp3 {
        out.map(s.rules().len());
        for rule in s.rules() {
            out.bulk(&rule.dest);
            out.array(3);
            out.int(rule.delta);
            out.simple(rule.agg.name().to_uppercase().as_bytes());
            out.int(rule.align);
        }
    } else {
        out.array(s.rules().len());
        for rule in s.rules() {
            out.array(4);
            out.bulk(&rule.dest);
            out.int(rule.delta);
            out.simple(rule.agg.name().to_uppercase().as_bytes());
            out.int(rule.align);
        }
    }
    out.simple(b"ignoreMaxTimeDiff");
    out.int(ignore_time);
    out.simple(b"ignoreMaxValDiff");
    // A plain double and not the shortest digits a sample value gets, so half a
    // degree is 0.5 here and 5E-1 out of `TS.GET`. That is two different reply
    // helpers inside one module rather than a decision, and both are copied.
    out.double(ignore_value);
    Ok(())
}

/// Puts a sample in and writes the reply, which is where all four of the
/// commands that write samples end up.
///
/// The timestamp that comes back is not always the one that went in. A sample
/// close enough to the newest one to be uninteresting is dropped and the newest
/// timestamp is answered instead, which is how a client tells the two apart.
fn store(body: &mut TsBody, at: i64, value: f64, over: Option<Policy>, out: &mut Out) -> bool {
    match body.s.add(Sample::new(at, value), over) {
        Ok(when) => {
            out.int(when);
            // A sample too close to the newest one to be interesting was not
            // stored, so there is nothing for a rule to be told about either.
            when == at
        }
        Err(Refused::Old) => {
            out.error_line(b"ERR ", TOO_OLD);
            false
        }
        Err(Refused::Duplicate) => {
            out.error_line(b"ERR ", UPSERT);
            false
        }
    }
}

/// Writes one of the sentences above with the `ERR` the module's own helper puts
/// in front of it.
fn say(out: &mut Out, msg: &[u8]) -> Result<()> {
    out.error_line(b"ERR ", msg);
    Ok(())
}

/// What the option words on a command said, with `None` for the ones that were
/// not there at all. That last part is what `TS.ALTER` needs: it changes what
/// was named and nothing else.
#[derive(Debug, Default)]
struct Options {
    /// How far back to keep samples.
    retention: Option<i64>,
    /// How much room to give a chunk.
    chunk_bytes: Option<usize>,
    /// How to store them.
    encoding: Option<Encoding>,
    /// What to do about a repeated timestamp.
    policy: Option<Policy>,
    /// The name and value pairs the series can be found by.
    labels: Option<Vec<(Vec<u8>, Vec<u8>)>>,
    /// How close to the newest sample a reading has to be to be dropped.
    ignore: Option<(i64, f64)>,
}

/// What went wrong reading the option words.
#[derive(Debug)]
enum Bad {
    /// A sentence the module writes through its own helper, which puts `ERR` in
    /// front of it.
    Said(&'static [u8]),
    /// One it writes straight to the client with nothing in front of it.
    Bare(&'static [u8]),
    /// The arity reply, which is what `ENCODING` with nothing behind it gets
    /// where every other keyword in the same spot gets a sentence.
    Arity,
}

/// Writes whatever `bad` says, in the shape the module says it in.
fn said(bad: Bad, name: &'static str, out: &mut Out) -> Result<()> {
    match bad {
        Bad::Said(msg) => say(out, msg),
        Bad::Bare(msg) => {
            out.error(msg);
            Ok(())
        }
        Bad::Arity => Err(args::wrong_arity(name)),
    }
}

/// Reads every option word out of a command, in the order the module reads them,
/// which is the order their errors come out in.
fn options(args: &Args<'_>) -> core::result::Result<Options, Bad> {
    let mut opts = Options::default();

    if let Some(at) = find(args, b"LABELS") {
        let first = at + 1;
        let mut labels = Vec::new();
        // Everything to the end, in pairs, with a trailing odd word dropped.
        for i in 0..args.len().saturating_sub(first) / 2 {
            let name = args.get(first + i * 2);
            let value = args.get(first + i * 2 + 1);
            // A label value has to be usable inside a filter expression, and the
            // three characters a filter is written with would make it
            // unreadable, so they are refused where the label is set instead.
            let ok = !name.is_empty()
                && !value.is_empty()
                && !value.iter().any(|b| matches!(b, b'(' | b')' | b','));
            if !ok {
                return Err(Bad::Said(BAD_LABELS));
            }
            labels.push((name.to_vec(), value.to_vec()));
        }
        opts.labels = Some(labels);
    }

    if let Some(at) = find(args, b"RETENTION") {
        let Some(n) = args.opt(at + 1).and_then(parse_i64) else {
            return Err(Bad::Said(BAD_RETENTION));
        };
        if n < 0 {
            return Err(Bad::Bare(BARE_RETENTION));
        }
        opts.retention = Some(n);
    }

    if let Some(at) = find(args, b"CHUNK_SIZE") {
        let Some(n) = args.opt(at + 1).and_then(parse_i64) else {
            return Err(Bad::Said(BAD_CHUNK));
        };
        // The range check runs first, so by the time the size is looked at as a
        // count of bytes it is known to be positive.
        if !(CHUNK_MIN..=CHUNK_MAX).contains(&n) || !(n as usize).is_multiple_of(8) {
            return Err(Bad::Said(CHUNK_RANGE));
        }
        opts.chunk_bytes = Some(n as usize);
    }

    if let Some(at) = find(args, b"ENCODING") {
        let Some(word) = args.opt(at + 1) else {
            return Err(Bad::Arity);
        };
        opts.encoding = Some(if args::is(word, b"uncompressed") {
            Encoding::Uncompressed
        } else if args::is(word, b"compressed") {
            Encoding::Compressed
        } else {
            return Err(Bad::Said(BAD_ENCODING));
        });
    }

    if let Some(at) = find(args, b"DUPLICATE_POLICY") {
        opts.policy = Some(policy_at(args, at)?);
    }

    if let Some(at) = find(args, b"IGNORE") {
        let time = args.opt(at + 1).and_then(parse_i64);
        let value = args.opt(at + 2).and_then(parse_f64);
        let (Some(time), Some(value)) = (time, value) else {
            return Err(Bad::Said(BAD_IGNORE));
        };
        if time < 0 || value < 0.0 {
            return Err(Bad::Said(NEGATIVE_IGNORE));
        }
        opts.ignore = Some((time, value));
    }

    Ok(opts)
}

/// The policy named just after `at`.
fn policy_at(args: &Args<'_>, at: usize) -> core::result::Result<Policy, Bad> {
    let Some(word) = args.opt(at + 1) else {
        return Err(Bad::Said(BAD_POLICY));
    };
    Policy::parse(word).ok_or(Bad::Said(UNKNOWN_POLICY))
}

/// Puts whatever was named onto a series and leaves the rest alone.
fn apply(s: &mut Series, opts: Options) {
    if let Some(n) = opts.retention {
        s.set_retention(n);
    }
    if let Some(n) = opts.chunk_bytes {
        s.set_chunk_bytes(n);
    }
    if let Some(encoding) = opts.encoding {
        s.set_encoding(encoding);
    }
    if let Some(policy) = opts.policy {
        s.set_policy(policy);
    }
    if let Some(labels) = opts.labels {
        s.set_labels(labels);
    }
    if let Some((time, value)) = opts.ignore {
        s.set_ignore(time, value);
    }
}

/// Where `word` first appears in the command, past the command name.
///
/// The module looks for each keyword across the whole argument list rather than
/// walking it once, which is why the key itself can be read as a keyword. See
/// the note at the top of the file.
fn find(args: &Args<'_>, word: &[u8]) -> Option<usize> {
    find_from(args, 1, word)
}

/// The same, starting at `from`, which is what the two increments need so that a
/// key called `TIMESTAMP` stays a key.
fn find_from(args: &Args<'_>, from: usize, word: &[u8]) -> Option<usize> {
    (from..args.len()).find(|&i| args::is(args.get(i), word))
}

/// One end of a `TS.DEL` span, which is a timestamp that is not below zero or
/// the one character standing for as far as that end goes.
fn span(arg: &[u8], open: &[u8], edge: i64) -> Option<i64> {
    if arg == open {
        return Some(edge);
    }
    parse_i64(arg).filter(|&n| n >= 0)
}

/// A timestamp argument, which is a whole number of milliseconds or a star for
/// whenever the server thinks it is now.
fn moment(db: &Db, arg: &[u8]) -> core::result::Result<i64, &'static [u8]> {
    if arg == b"*" {
        return Ok(now(db));
    }
    let Some(at) = parse_i64(arg) else {
        return Err(BAD_TIMESTAMP);
    };
    if at < 0 {
        return Err(NEGATIVE_TIMESTAMP);
    }
    Ok(at)
}

/// The server's clock, as a timestamp.
fn now(db: &Db) -> i64 {
    db.now_ms() as i64
}

/// A sample value, in the grammar the module's own reader accepts.
///
/// That is an optional minus, at least one digit, an optional fraction with at
/// least one digit and an optional exponent with at least one digit, or one of
/// the three spellings of a NaN. No leading plus, no bare fraction, no space at
/// either end, and no infinity: a number too large to hold is refused rather
/// than stored as one. The increment on `TS.INCRBY` goes through the ordinary
/// number reader instead and takes all of those, which is not a distinction
/// anyone designed.
fn number(arg: &[u8]) -> Option<f64> {
    if nan_word(arg) {
        return Some(f64::NAN);
    }
    let mut at = usize::from(arg.first() == Some(&b'-'));
    if !digits(arg, &mut at) {
        return None;
    }
    if arg.get(at) == Some(&b'.') {
        at += 1;
        if !digits(arg, &mut at) {
            return None;
        }
    }
    if matches!(arg.get(at), Some(b'e' | b'E')) {
        at += 1;
        if matches!(arg.get(at), Some(b'+' | b'-')) {
            at += 1;
        }
        if !digits(arg, &mut at) {
            return None;
        }
    }
    if at != arg.len() {
        return None;
    }
    parse_f64(arg).filter(|value| value.is_finite())
}

/// Steps `at` over a run of digits, and answers whether there was one.
fn digits(arg: &[u8], at: &mut usize) -> bool {
    let start = *at;
    while arg.get(*at).is_some_and(u8::is_ascii_digit) {
        *at += 1;
    }
    *at > start
}

/// Whether `arg` is one of the three ways the module spells a reading that is
/// not a number.
fn nan_word(arg: &[u8]) -> bool {
    match arg.len() {
        3 => arg.eq_ignore_ascii_case(b"nan"),
        4 => arg.eq_ignore_ascii_case(b"-nan") || arg.eq_ignore_ascii_case(b"+nan"),
        _ => false,
    }
}

/// A sample value on the way out: the shortest digits that read back as the same
/// number, as a simple string on RESP2 and a double on RESP3.
fn value(out: &mut Out, d: f64) {
    if out.proto().is_resp3() {
        out.double(d);
        return;
    }
    let mut buf = [0u8; DOUBLE_MAX];
    out.simple(write_dragonbox(&mut buf, d));
}

/// The wrong kind of key, worded the way the module words it.
///
/// The keyspace answers its own `WRONGTYPE` for a key holding a native type and
/// the downcast answers for a key holding another module's value, and both of
/// those reach a client as the same sentence, so both come through here.
fn wrong_kind() -> Error {
    Error::new(Code::Invalid, WRONG_KIND)
}

/// The series under `key` for writing, or `None` if the key is not there.
///
/// The stripe comes in rather than being worked out here, because the answer
/// borrows out of it and a borrow cannot outlive the guard it came from. The
/// caller decides how long it holds, which is what a command that writes
/// through a rule into another key needs: it lets go of the source before it
/// reaches for the destination.
fn write<'k>(stripe: &'k mut Keyspace, key: &[u8]) -> Result<Option<&'k mut TsBody>> {
    match stripe.foreign_mut(key) {
        Ok(Some(body)) => match body.downcast_mut::<TsBody>() {
            Some(body) => Ok(Some(body)),
            None => Err(wrong_kind()),
        },
        Ok(None) => Ok(None),
        Err(e) if e.code() == Code::WrongType => Err(wrong_kind()),
        Err(e) => Err(e),
    }
}

/// The same again, with the server's own bare `WRONGTYPE` for a key holding
/// something else rather than the module's prefixed sentence, which is what
/// `TS.READ` answers and the only place in the family that does.
fn bare_read<'k>(stripe: &'k mut Keyspace, key: &[u8]) -> Result<Option<&'k TsBody>> {
    match stripe.foreign(key)? {
        Some(body) => match body.downcast_ref::<TsBody>() {
            Some(body) => Ok(Some(body)),
            None => Err(yo_kv::keyspace::wrong_type()),
        },
        None => Ok(None),
    }
}

/// The same, for reading.
fn read<'k>(stripe: &'k mut Keyspace, key: &[u8]) -> Result<Option<&'k TsBody>> {
    match stripe.foreign(key) {
        Ok(Some(body)) => match body.downcast_ref::<TsBody>() {
            Some(body) => Ok(Some(body)),
            None => Err(wrong_kind()),
        },
        Ok(None) => Ok(None),
        Err(e) if e.code() == Code::WrongType => Err(wrong_kind()),
        Err(e) => Err(e),
    }
}