use super::args::{self, Args};
use super::scan;
use super::table::Spec;
use crate::reply::Out;
use yo_common::{Code, Error, Held, Result, glob_matches};
use yo_kv::rdb::Bad;
use yo_kv::sort::Sort;
use yo_kv::{Applied, Ask, Cond, Db, Holds, Keyspace, Kind, MAX_AT, Moved};
const SECOND: i64 = 1000;
const DB_OUT_OF_RANGE: &str = "DB index is out of range";
const SAME_OBJECT: &str = "source and destination objects are the same";
const BUSY_KEY: &[u8] = b"BUSYKEY Target key name already exists.";
const BAD_TTL: &str = "Invalid TTL value, must be >= 0";
const BAD_IDLETIME: &str = "Invalid IDLETIME value, must be >= 0";
const BAD_FREQ: &str = "Invalid FREQ value, must be >= 0 and <= 255";
const BAD_FOOTER: &str = "DUMP payload version or checksum are wrong";
const BAD_PAYLOAD: &str = "Bad data format";
const NX_WITH_OTHERS: &str = "NX and XX, GT or LT options at the same time are not compatible";
const GT_WITH_LT: &str = "GT and LT options at the same time are not compatible";
const NOT_LFU: &str = "An LFU maxmemory policy is not selected, access frequency not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.";
const IS_LFU: &str = "An LFU maxmemory policy is selected, idle time not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.";
const OBJECT_HELP: &[&str] = &[
"OBJECT <subcommand> [<arg> [value] [opt] ...]. Subcommands are:",
"ENCODING <key>",
" Return the kind of internal representation used in order to store the value",
" associated with a <key>.",
"FREQ <key>",
" Return the access frequency index of the <key>. The returned integer is",
" proportional to the logarithm of the real access frequency.",
"IDLETIME <key>",
" Return the idle time of the <key>, that is the approximated number of",
" seconds elapsed since the last access to the value.",
"REFCOUNT <key>",
" Return the number of references of the value associated with the key.",
"HELP",
" Print this help.",
];
pub(super) fn execute(
dbs: &[Db],
at: usize,
spec: &Spec,
args: Args<'_>,
out: &mut Out,
) -> Result<()> {
match spec.name {
"copy" => return copy(dbs, at, args, out),
"move" => return move_key(dbs, at, args, out),
_ => {}
}
let db = &dbs[at];
match spec.name {
"del" | "unlink" => {
let mut gone = 0i64;
for i in 1..args.len() {
let key = args.get(i);
if db.hold(key).del(key) {
gone += 1;
}
}
out.int(gone);
}
"exists" => {
let mut found = 0i64;
for i in 1..args.len() {
let key = args.get(i);
if db.hold(key).exists(key) {
found += 1;
}
}
out.int(found);
}
"type" => {
let key = args.get(1);
let name = match db.hold(key).type_name(key) {
Some(name) => name.as_bytes(),
None => &b"none"[..],
};
out.simple(name);
}
"touch" => {
let mut hit = 0i64;
for i in 1..args.len() {
let key = args.get(i);
hit += db.hold(key).touch(core::iter::once(key)) as i64;
}
out.int(hit);
}
"rename" | "renamenx" => rename(db, spec.name, args, out)?,
"expire" | "pexpire" | "expireat" | "pexpireat" => expire(db, spec.name, args, out)?,
"persist" => {
let key = args.get(1);
out.int(i64::from(db.hold(key).persist(key)));
}
"ttl" | "pttl" | "expiretime" | "pexpiretime" => ask(db, spec.name, args, out),
"object" => object(db, args, out)?,
"scan" => scan(db, args, out)?,
"keys" => keys(db, args.get(1), out),
"sort" | "sort_ro" => sort(db, spec.name, args, out)?,
"dump" => {
let key = args.get(1);
let mut stripe = db.hold(key);
if is_foreign(&mut stripe, key) {
return Err(no_dump(&mut stripe, key));
}
match stripe.dump(key) {
Some(payload) => out.bulk(&payload),
None => out.nil(),
}
}
"restore" => restore(db, args, out)?,
"randomkey" => {
if !db.random_key(|key| out.bulk(key)) {
out.nil();
}
}
other => unreachable!("keyspace command with no body: {other}"),
}
Ok(())
}
fn scan(db: &Db, args: Args<'_>, out: &mut Out) -> Result<()> {
let cursor = scan::parse_cursor(args.get(1))?;
let mut pattern = None;
let mut count = scan::COUNT;
let mut ty = None;
let mut impossible = false;
let mut i = 2;
while i < args.len() {
let rest = args.len() - i;
if args::is(args.get(i), b"match") && rest >= 2 {
pattern = Some(args.get(i + 1));
} else if args::is(args.get(i), b"count") && rest >= 2 {
count = match args.int(i + 1)? {
n if n >= 1 => usize::try_from(n).unwrap_or(usize::MAX),
_ => return Err(args::syntax()),
};
} else if args::is(args.get(i), b"type") && rest >= 2 {
match kind_named(args.get(i + 1)) {
Some(k) => ty = Some(k),
None => impossible = true,
}
} else {
return Err(args::syntax());
}
i += 2;
}
scan::reply(out, |out| {
let mut n = 0;
let next = db.scan(cursor, count, ty, |key| {
if impossible || pattern.is_some_and(|p| !glob_matches(p, key)) {
return;
}
out.bulk(key);
n += 1;
});
Ok((next, n))
})
}
fn keys(db: &Db, pattern: &[u8], out: &mut Out) {
let at = out.len();
let mut n = 0;
db.keys(|key| {
if glob_matches(pattern, key) {
out.bulk(key);
n += 1;
}
});
out.close_array(at, n);
}
fn kind_named(arg: &[u8]) -> Option<Kind> {
[
Kind::String,
Kind::Hash,
Kind::Set,
Kind::Zset,
Kind::List,
Kind::Stream,
]
.into_iter()
.find(|kind| args::is(arg, kind.name().as_bytes()))
}
const MODULE_KEY: &str = "not supported for this module key";
fn is_foreign(db: &mut Keyspace, key: &[u8]) -> bool {
db.kind_of(key) == Some(Kind::Foreign)
}
fn no_copy(db: &mut Keyspace, key: &[u8]) -> Error {
match db.type_name(key) {
Some("graph") => Error::new(Code::Unsupported, "COPY is not supported for a graph"),
_ => Error::new(Code::Unsupported, MODULE_KEY),
}
}
fn no_dump(db: &mut Keyspace, key: &[u8]) -> Error {
match db.type_name(key) {
Some("graph") => Error::new(Code::Unsupported, "DUMP is not supported for a graph"),
_ => Error::new(
Code::Unsupported,
"DUMP is not supported for this module key",
),
}
}
fn rename(db: &Db, name: &str, args: Args<'_>, out: &mut Out) -> Result<()> {
let nx = name == "renamenx";
let (src, dst) = (args.get(1), args.get(2));
let (from, to) = (db.stripe_of(src), db.stripe_of(dst));
let done = if from == to {
db.hold_stripe(from).rename(src, dst, nx)
} else {
let mut held = db.hold_many([from, to].into_iter());
rename_across(&mut held, from, to, src, dst, nx)
};
let done = done.found()?;
if nx {
out.int(i64::from(done == Moved::Ok));
} else {
out.ok();
}
Ok(())
}
fn rename_across(
held: &mut Holds<'_>,
from: usize,
to: usize,
src: &[u8],
dst: &[u8],
nx: bool,
) -> Moved {
if !held.stripe_mut(from).exists(src) {
return Moved::Missing;
}
if nx && held.stripe_mut(to).exists(dst) {
return Moved::Taken;
}
let rec = held
.stripe_mut(from)
.take(src)
.expect("the source was live a line ago");
held.stripe_mut(to).import(dst, rec);
Moved::Ok
}
fn copy(dbs: &[Db], at: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
let (src, dst) = (args.get(1), args.get(2));
let mut into = at;
let mut replace = false;
let mut i = 3;
while i < args.len() {
let arg = args.get(i);
if args::is(arg, b"replace") {
replace = true;
i += 1;
continue;
}
if !args::is(arg, b"db") || i + 1 >= args.len() {
return Err(args::syntax());
}
let n = args.int(i + 1)?;
into = usize::try_from(n)
.ok()
.filter(|n| *n < dbs.len())
.ok_or_else(|| Error::new(Code::Invalid, DB_OUT_OF_RANGE))?;
i += 2;
}
if into == at && src == dst {
return Err(Error::new(Code::Invalid, SAME_OBJECT));
}
let (from, to) = (spot(dbs, at, src), spot(dbs, into, dst));
let done = match hold_both(dbs, from, to) {
Both::One(mut ks) => match ks.copy(src, dst, replace) {
Moved::Unsupported => return Err(no_copy(&mut ks, src)),
done => done,
},
Both::Two(mut from, mut to) => {
if !replace && to.exists(dst) {
out.int(0);
return Ok(());
}
if is_foreign(&mut from, src) {
return Err(no_copy(&mut from, src));
}
let Some(rec) = from.export(src) else {
out.int(0);
return Ok(());
};
to.import(dst, rec);
Moved::Ok
}
};
out.int(i64::from(done == Moved::Ok));
Ok(())
}
type Spot = (usize, usize);
fn spot(dbs: &[Db], db: usize, key: &[u8]) -> Spot {
(db, dbs[db].stripe_of(key))
}
enum Both<'d> {
One(Held<'d, Keyspace>),
Two(Held<'d, Keyspace>, Held<'d, Keyspace>),
}
fn hold_both(dbs: &[Db], from: Spot, to: Spot) -> Both<'_> {
if from == to {
return Both::One(dbs[from.0].hold_stripe(from.1));
}
if from < to {
let source = dbs[from.0].hold_stripe(from.1);
let dest = dbs[to.0].hold_stripe(to.1);
Both::Two(source, dest)
} else {
let dest = dbs[to.0].hold_stripe(to.1);
let source = dbs[from.0].hold_stripe(from.1);
Both::Two(source, dest)
}
}
fn restore(db: &Db, args: Args<'_>, out: &mut Out) -> Result<()> {
let key = args.get(1);
let mut db = db.hold(key);
let mut replace = false;
let mut absolute = false;
let mut idle = -1i64;
let mut freq = -1i64;
let mut i = 4;
while i < args.len() {
let arg = args.get(i);
let more = args.len() - i - 1;
if args::is(arg, b"replace") {
replace = true;
} else if args::is(arg, b"absttl") {
absolute = true;
} else if args::is(arg, b"idletime") && more >= 1 && freq == -1 {
idle = args.int(i + 1)?;
if idle < 0 {
return Err(Error::new(Code::Invalid, BAD_IDLETIME));
}
i += 1;
} else if args::is(arg, b"freq") && more >= 1 && idle == -1 {
freq = args.int(i + 1)?;
if !(0..=255).contains(&freq) {
return Err(Error::new(Code::Invalid, BAD_FREQ));
}
i += 1;
} else {
return Err(args::syntax());
}
i += 1;
}
if !replace && db.exists(key) {
out.error(BUSY_KEY);
return Ok(());
}
let ttl = args.int(2)?;
if ttl < 0 {
return Err(Error::new(Code::Invalid, BAD_TTL));
}
let now = db.clock().now_ms();
let ttl = ttl as u64;
let expire_at = match ttl {
0 => None,
_ if absolute => Some(ttl.min(MAX_AT)),
_ => Some(now.saturating_add(ttl).min(MAX_AT)),
};
let _ = (idle, freq);
match db.restore(key, args.get(3), expire_at, replace) {
Ok(_) => out.ok(),
Err(Bad::Footer) => return Err(Error::new(Code::Invalid, BAD_FOOTER)),
Err(Bad::Format) => return Err(Error::new(Code::Invalid, BAD_PAYLOAD)),
}
Ok(())
}
fn move_key(dbs: &[Db], at: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
let key = args.get(1);
let n = args.int(2)?;
let into = usize::try_from(n)
.ok()
.filter(|n| *n < dbs.len())
.ok_or_else(|| Error::new(Code::Invalid, DB_OUT_OF_RANGE))?;
if into == at {
return Err(Error::new(Code::Invalid, SAME_OBJECT));
}
let (from, to) = (spot(dbs, at, key), spot(dbs, into, key));
let Both::Two(mut from, mut to) = hold_both(dbs, from, to) else {
unreachable!("a move into the database the key is already in was refused above")
};
if to.exists(key) {
out.int(0);
return Ok(());
}
let Some(rec) = from.take(key) else {
out.int(0);
return Ok(());
};
to.import(key, rec);
out.int(1);
Ok(())
}
fn sort(db: &Db, name: &str, args: Args<'_>, out: &mut Out) -> Result<()> {
let read_only = name == "sort_ro";
let key = args.get(1);
let mut opts = Sort::default();
let mut get: Vec<&[u8]> = Vec::new();
let mut store: Option<&[u8]> = None;
let mut i = 2;
while i < args.len() {
let arg = args.get(i);
let rest = args.len() - i;
if args::is(arg, b"asc") {
opts.desc = false;
} else if args::is(arg, b"desc") {
opts.desc = true;
} else if args::is(arg, b"alpha") {
opts.alpha = true;
} else if args::is(arg, b"by") && rest >= 2 {
opts.by = Some(args.get(i + 1));
i += 2;
continue;
} else if args::is(arg, b"get") && rest >= 2 {
get.push(args.get(i + 1));
i += 2;
continue;
} else if args::is(arg, b"limit") && rest >= 3 {
opts.limit = Some((args.int(i + 1)?, args.int(i + 2)?));
i += 3;
continue;
} else if args::is(arg, b"store") && rest >= 2 && !read_only {
store = Some(args.get(i + 1));
i += 2;
continue;
} else {
return Err(args::syntax());
}
i += 1;
}
opts.get = &get;
match store {
Some(dst) => out.int(i64::try_from(db.sort_store(key, dst, &opts)?).unwrap_or(i64::MAX)),
None => {
let rows = db.sort(key, &opts)?;
out.array(rows.len());
for row in rows {
match row {
Some(v) => out.bulk(&v),
None => out.nil(),
}
}
}
}
Ok(())
}
fn expire(db: &Db, name: &str, args: Args<'_>, out: &mut Out) -> Result<()> {
let mut db = db.hold(args.get(1));
let relative = matches!(name, "expire" | "pexpire");
let scale = if matches!(name, "pexpire" | "pexpireat") {
1
} else {
SECOND
};
let at = moment(args.int(2)?, scale, relative, name, db.clock().now_ms())?;
let cond = condition(args)?;
out.int(match db.expire(args.get(1), at, cond) {
Applied::Missing | Applied::NotMet => 0,
Applied::Ok | Applied::Deleted => 1,
});
Ok(())
}
fn moment(by: i64, scale: i64, relative: bool, name: &str, now: u64) -> Result<u64> {
let ms = by
.checked_mul(scale)
.and_then(|ms| {
if relative {
ms.checked_add(now as i64)
} else {
Some(ms)
}
})
.ok_or_else(|| {
Error::new(
Code::Invalid,
format!("invalid expire time in '{name}' command"),
)
})?;
Ok(ms.clamp(0, MAX_AT as i64) as u64)
}
fn condition(args: Args<'_>) -> Result<Cond> {
let (mut nx, mut xx, mut gt, mut lt) = (false, false, false, false);
for i in 3..args.len() {
let arg = args.get(i);
match arg {
a if args::is(a, b"nx") => nx = true,
a if args::is(a, b"xx") => xx = true,
a if args::is(a, b"gt") => gt = true,
a if args::is(a, b"lt") => lt = true,
_ => {
return Err(Error::new(
Code::Invalid,
format!("Unsupported option {}", String::from_utf8_lossy(arg)),
));
}
}
}
if nx && (xx || gt || lt) {
return Err(Error::new(Code::Invalid, NX_WITH_OTHERS));
}
if gt && lt {
return Err(Error::new(Code::Invalid, GT_WITH_LT));
}
Ok(match (nx, xx, gt, lt) {
(true, ..) => Cond::NotSet,
(_, _, true, _) => Cond::Greater,
(_, true, _, true) => Cond::LessAndSet,
(_, _, _, true) => Cond::Less,
(_, true, ..) => Cond::AlreadySet,
_ => Cond::Always,
})
}
fn ask(db: &Db, name: &str, args: Args<'_>, out: &mut Out) {
let key = args.get(1);
let mut db = db.hold(key);
let now = db.clock().now_ms();
let asked = db.deadline_of(key);
let millis = name == "pttl" || name == "pexpiretime";
let ms = if name == "ttl" || name == "pttl" {
asked.remaining_ms(now)
} else {
match asked {
Ask::Missing => -2,
Ask::NoDeadline => -1,
Ask::At(at) => at as i64,
}
};
out.int(if millis || ms < 0 {
ms
} else {
(ms + SECOND / 2) / SECOND
});
}
fn object(db: &Db, args: Args<'_>, out: &mut Out) -> Result<()> {
let sub = args.get(1);
if args::is(sub, b"help") {
if args.len() != 2 {
return Err(args::unknown_subcommand(sub, "OBJECT"));
}
out.array(OBJECT_HELP.len());
for line in OBJECT_HELP {
out.simple(line.as_bytes());
}
return Ok(());
}
let named = ["encoding", "refcount", "idletime", "freq"]
.into_iter()
.find(|n| args::is(sub, n.as_bytes()));
let Some(named) = named else {
return Err(args::unknown_subcommand(sub, "OBJECT"));
};
if args.len() != 3 {
return Err(args::wrong_arity_sub("object", named));
}
let key = args.get(2);
let mut db = db.hold(key);
if !db.exists(key) {
out.nil();
return Ok(());
}
match named {
"encoding" => {
let name = db
.encoding_name(key)
.expect("the key is there, so it has an encoding");
out.bulk(name.as_bytes());
}
"refcount" => out.int(1),
"idletime" => {
if !db.policy().is_clock() {
return Err(Error::new(Code::Unsupported, IS_LFU));
}
let idle = db.idle_secs(key).expect("the key is there");
out.int(i64::try_from(idle).unwrap_or(i64::MAX));
}
"freq" => {
if !db.policy().is_lfu() {
return Err(Error::new(Code::Unsupported, NOT_LFU));
}
let freq = db.freq(key).expect("the key is there");
out.int(i64::from(freq));
}
other => unreachable!("no body for object {other}"),
}
Ok(())
}