Apache Cassandra

use cassandra.qweruse cassandra_connect, cassandra_close from cassandra      // Quillo 0.6: only these

hex = "0.4"serde_json = { version = "1", features = ["preserve_order"] }

// A Cassandra cluster from a program: statements prepared and run with values of each CQL type, one// row of every type read back, a table read a page at a time, a batch, and what the server's// errors say.////   CASSANDRA_HOST=127.0.0.1:9042 CASSANDRA_USER=cassandra CASSANDRA_PASSWORD=cassandra quillo run cassandra.qwe//   then open http://localhost:8080/connect, /write, /types, /pages, /batch and /failures//// It wants a keyspace `shop`, with tables as the routes use them.use "cassandra.qwer"let opened = cassandra_connect("main", { keyspace: "shop", consistency: "LOCAL_QUORUM", page_size: 3 })route GET /connect {    let c = opened?    { connected: c.connected, versions: c.supported.CQL_VERSION, keyspace: cassandra_query("main", "SELECT keyspace_name FROM system.local", null, {})?.columns[0].type, no_password: is_err(cassandra_connect("second", { keyspace: "bad name" })) }}route GET /write {    opened?    let insert = "INSERT INTO orders (id, customer, total, placed, tags) VALUES (?, ?, ?, ?, ?)"    let id = "123e4567-e89b-12d3-a456-426614174000"    cassandra_execute("main", insert, [id, "ada", 42.5, "2026-11-01T09:30:00.250Z", ["new", "gift"]], {})?    cassandra_execute("main", insert, { id: "123e4567-e89b-12d3-a456-426614174001", customer: "bob", total: 7.0, placed: 1730000000000, tags: [] }, {})?    let got = cassandra_execute("main", "SELECT id, customer, total, placed, tags FROM orders WHERE id = ?", [id], {})?    { row: got.rows[0], columns: map(got.columns, (c) => c.type), again: len(cassandra_execute("main", "SELECT id, customer, total, placed, tags FROM orders WHERE id = ?", [id], {})?.rows) }}route GET /types {    opened?    let got = cassandra_query("main", "SELECT * FROM every_type", null, {})?    { row: got.rows[0], types: map(got.columns, (c) => c.type), nulls: got.rows[1].a_text }}route GET /bind {    opened?    let cql = "INSERT INTO bound (k, big, small, tiny, dec, vi, ip, day, flag, blobby, scores, pair) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"    cassandra_execute("main", cql, ["k1", 9007199254740993, 300, -5, "-123.450", "-170141183460469231731687303715884105728", "2001:db8::1", "2026-11-01", true, "0xcafe", { a: 1, b: 2 }, [7, "x"]], {})?    cassandra_execute("main", "SELECT * FROM bound WHERE k = ?", ["k1"], {})?.rows[0]}route GET /pages {    opened?    let one = cassandra_execute("main", "SELECT n FROM events WHERE day = ?", ["2026-11-01"], { page_size: 4 })?    let two = cassandra_execute("main", "SELECT n FROM events WHERE day = ?", ["2026-11-01"], { page_size: 4, paging_state: one.paging_state })?    let all = cassandra_all("main", "SELECT n FROM events WHERE day = ?", ["2026-11-01"], { page_size: 4 })?    let capped = cassandra_all("main", "SELECT n FROM events WHERE day = ?", ["2026-11-01"], { page_size: 4, max_rows: 6 })?    { first: map(one.rows, (r) => r.n), more: one.has_more, second: map(two.rows, (r) => r.n), all: len(all.rows), pages: all.pages, capped: len(capped.rows) }}// The stand-in forgets its prepared statements after a batch, as a restarted node does: the read// after it is answered Unprepared, and prepared again without the program seeing.route GET /batch {    opened?    cassandra_execute("main", "SELECT balance FROM accounts WHERE name = ?", ["ada"], {})?    let ok = cassandra_batch("main", [        { cql: "INSERT INTO accounts (name, balance) VALUES (?, ?)", values: ["ada", 10] },        { cql: "INSERT INTO accounts (name, balance) VALUES (?, ?)", values: ["bob", 20] },    ], { consistency: "QUORUM" })?    {        ok: ok,        rows: cassandra_execute("main", "SELECT balance FROM accounts WHERE name = ?", ["bob"], {})?.rows,        warnings: cassandra_warnings("main")?,        evicted: cassandra_execute("main", "SELECT balance FROM accounts WHERE name = ?", ["ada"], {})?.rows,    }}route GET /failures {    opened?    {        syntax: cassandra_query("main", "SELEKT nothing", null, {}).error,        missing: cassandra_execute("main", "SELECT * FROM absent WHERE k = ?", ["x"], {}).error,        few_values: cassandra_execute("main", "INSERT INTO accounts (name, balance) VALUES (?, ?)", ["only one"], {}).error,        wrong_type: cassandra_execute("main", "INSERT INTO accounts (name, balance) VALUES (?, ?)", ["a", "not a number"], {}).error,        level: cassandra_query("main", "SELECT 1", null, { consistency: "MOST" }).error,        unavailable: cassandra_query("main", "SELECT unavailable", null, {}).error,        timeout: cassandra_query("main", "SELECT timeout", null, {}).error,        unknown_connection: cassandra_query("nobody", "SELECT 1", null, {}).error,        batch_kind: cassandra_batch("main", [], { kind: "fast" }).error,    }}

//! Apache Cassandra and ScyllaDB by the CQL binary protocol v4, by hand: STARTUP and password auth, plain and prepared statements with an Unprepared answer re-prepared, batches, result rows read by type (collections, tuples, UDTs, decimals, UUIDs, times), paging state followed, and server errors with their codes.//! version: 1.0.0//!//! [dependencies]//! hex = "0.4"//! serde_json = { version = "1", features = ["preserve_order"] }use serde_json::{json, Map, Value};use std::collections::HashMap;use std::io::{Read, Write};use std::net::{IpAddr, TcpStream, ToSocketAddrs};use std::sync::{Arc, Mutex};use std::time::Duration;const STARTUP: u8 = 1;const READY: u8 = 2;const AUTHENTICATE: u8 = 3;const OPTIONS: u8 = 5;const SUPPORTED: u8 = 6;const QUERY: u8 = 7;const RESULT: u8 = 8;const PREPARE: u8 = 9;const EXECUTE: u8 = 10;const BATCH: u8 = 13;const AUTH_CHALLENGE: u8 = 14;const AUTH_RESPONSE: u8 = 15;const AUTH_SUCCESS: u8 = 16;// ---- Reading a body ----struct Rd<'a>(&'a [u8]);impl<'a> Rd<'a> {    fn take(&mut self, n: usize) -> Result<&'a [u8], String> {        if self.0.len() < n {            return Err("a cut-off CQL message".into());        }        let (a, b) = self.0.split_at(n);        self.0 = b;        Ok(a)    }    fn u8(&mut self) -> Result<u8, String> {        Ok(self.take(1)?[0])    }    fn u16(&mut self) -> Result<u16, String> {        let b = self.take(2)?;        Ok(u16::from_be_bytes([b[0], b[1]]))    }    fn i32(&mut self) -> Result<i32, String> {        let b = self.take(4)?;        Ok(i32::from_be_bytes([b[0], b[1], b[2], b[3]]))    }    fn string(&mut self) -> Result<String, String> {        let n = self.u16()? as usize;        String::from_utf8(self.take(n)?.to_vec()).map_err(|_| "a CQL string that is not UTF-8".to_string())    }    fn long_string(&mut self) -> Result<String, String> {        let n = self.i32()?.max(0) as usize;        String::from_utf8(self.take(n)?.to_vec()).map_err(|_| "a CQL string that is not UTF-8".to_string())    }    fn bytes(&mut self) -> Result<Option<&'a [u8]>, String> {        let n = self.i32()?;        if n < 0 { Ok(None) } else { Ok(Some(self.take(n as usize)?)) }    }    fn short_bytes(&mut self) -> Result<&'a [u8], String> {        let n = self.u16()? as usize;        self.take(n)    }    fn string_list(&mut self) -> Result<Vec<String>, String> {        (0..self.u16()?).map(|_| self.string()).collect()    }}// ---- Types ----#[derive(Clone, Debug)]enum Ty {    Plain(u16),    Custom(String),    List(Box<Ty>),    Set(Box<Ty>),    Map(Box<Ty>, Box<Ty>),    Udt(Vec<(String, Ty)>),    Tuple(Vec<Ty>),}fn read_type(r: &mut Rd) -> Result<Ty, String> {    let id = r.u16()?;    Ok(match id {        0 => Ty::Custom(r.string()?),        0x20 => Ty::List(Box::new(read_type(r)?)),        0x22 => Ty::Set(Box::new(read_type(r)?)),        0x21 => {            let k = read_type(r)?;            Ty::Map(Box::new(k), Box::new(read_type(r)?))        }        0x30 => {            r.string()?;            r.string()?;            let n = r.u16()?;            let mut f = vec![];            for _ in 0..n {                let name = r.string()?;                f.push((name, read_type(r)?));            }            Ty::Udt(f)        }        0x31 => {            let n = r.u16()?;            Ty::Tuple((0..n).map(|_| read_type(r)).collect::<Result<_, _>>()?)        }        x => Ty::Plain(x),    })}fn type_name(t: &Ty) -> String {    match t {        Ty::Plain(x) => match x {            1 => "ascii", 2 => "bigint", 3 => "blob", 4 => "boolean", 5 => "counter", 6 => "decimal", 7 => "double", 8 => "float", 9 => "int",            0xb => "timestamp", 0xc => "uuid", 0xd => "varchar", 0xe => "varint", 0xf => "timeuuid", 0x10 => "inet", 0x11 => "date", 0x12 => "time", 0x13 => "smallint", 0x14 => "tinyint",            _ => "unknown",        }.to_string(),        Ty::Custom(n) => n.clone(),        Ty::List(e) => format!("list<{}>", type_name(e)),        Ty::Set(e) => format!("set<{}>", type_name(e)),        Ty::Map(k, v) => format!("map<{}, {}>", type_name(k), type_name(v)),        Ty::Udt(_) => "udt".to_string(),        Ty::Tuple(l) => format!("tuple<{}>", l.iter().map(type_name).collect::<Vec<_>>().join(", ")),    }}fn civil_iso(days: i64) -> String {    let z = days + 719468;    let era = z.div_euclid(146097);    let doe = z - era * 146097;    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);    let mp = (5 * doy + 2) / 153;    let d = doy - (153 * mp + 2) / 5 + 1;    let m = if mp < 10 { mp + 3 } else { mp - 9 };    format!("{:04}-{:02}-{:02}", yoe + era * 400 + if m <= 2 { 1 } else { 0 }, m, d)}fn days_of(s: &str) -> Option<i64> {    let p: Vec<i64> = s.get(..10)?.split('-').map(|x| x.parse().ok()).collect::<Option<_>>()?;    if p.len() != 3 {        return None;    }    let (y, m) = if p[1] > 2 { (p[0], p[1]) } else { (p[0] - 1, p[1] + 12) };    let era = y.div_euclid(400);    let yoe = y - era * 400;    Some(era * 146097 + yoe * 365 + yoe / 4 - yoe / 100 + (153 * (m - 3) + 2) / 5 + p[2] - 1 - 719468)}fn stamp(ms: i64) -> String {    let (d, r) = (ms.div_euclid(86_400_000), ms.rem_euclid(86_400_000));    format!("{}T{:02}:{:02}:{:02}.{:03}Z", civil_iso(d), r / 3_600_000, r / 60_000 % 60, r / 1000 % 60, r % 1000)}fn uuid_text(b: &[u8]) -> String {    let h = hex::encode(b);    format!("{}-{}-{}-{}-{}", &h[..8], &h[8..12], &h[12..16], &h[16..20], &h[20..])}/// A two's-complement big-endian integer of any length, as text.fn varint(b: &[u8]) -> String {    if b.is_empty() {        return "0".into();    }    let neg = b[0] & 0x80 != 0;    let mut mag: Vec<u8> = if neg { b.iter().map(|x| !x).collect() } else { b.to_vec() };    if neg {        for x in mag.iter_mut().rev() {            *x = x.wrapping_add(1);            if *x != 0 {                break;            }        }    }    let mut digits = vec![0u8];    for byte in mag {        let mut carry = byte as u32;        for d in digits.iter_mut() {            let v = *d as u32 * 256 + carry;            *d = (v % 10) as u8;            carry = v / 10;        }        while carry > 0 {            digits.push((carry % 10) as u8);            carry /= 10;        }    }    let t: String = digits.iter().rev().map(|d| (b'0' + d) as char).collect();    let t = t.trim_start_matches('0');    match (t.is_empty(), neg) {        (true, _) => "0".into(),        (_, true) => format!("-{}", t),        _ => t.to_string(),    }}/// Text of a decimal integer as two's complement.fn varint_bytes(text: &str) -> Result<Vec<u8>, String> {    let t = text.trim();    let (neg, digits) = t.strip_prefix('-').map_or((false, t), |r| (true, r));    if digits.is_empty() || !digits.bytes().all(|c| c.is_ascii_digit()) {        return Err(format!("`{}` is not a whole number", text));    }    let mut mag = vec![0u8];    for c in digits.bytes() {        let mut carry = (c - b'0') as u32;        for x in mag.iter_mut().rev() {            let v = *x as u32 * 10 + carry;            *x = v as u8;            carry = v >> 8;        }        if carry > 0 {            mag.insert(0, carry as u8);        }    }    if mag[0] & 0x80 != 0 {        mag.insert(0, 0);    }    if neg {        for x in mag.iter_mut() {            *x = !*x;        }        for x in mag.iter_mut().rev() {            *x = x.wrapping_add(1);            if *x != 0 {                break;            }        }    }    Ok(mag)}fn decode(t: &Ty, b: Option<&[u8]>) -> Result<Value, String> {    let Some(b) = b else { return Ok(Value::Null) };    Ok(match t {        Ty::Plain(x) => match x {            1 | 0xd => json!(String::from_utf8_lossy(b)),            2 | 5 => json!(i64::from_be_bytes(b.try_into().map_err(|_| "a bigint of the wrong size")?)),            3 => json!(format!("0x{}", hex::encode(b))),            4 => json!(b.first() == Some(&1)),            6 => {                if b.len() < 4 {                    return Err("a cut-off decimal".into());                }                let scale = i32::from_be_bytes([b[0], b[1], b[2], b[3]]);                let digits = varint(&b[4..]);                let (neg, d) = digits.strip_prefix('-').map_or((false, digits.as_str()), |r| (true, r));                let s = scale.max(0) as usize;                let padded = format!("{:0>width$}", d, width = s + 1);                let (a, c) = padded.split_at(padded.len() - s);                json!(format!("{}{}{}{}", if neg { "-" } else { "" }, a, if s > 0 { "." } else { "" }, c))            }            7 => json!(f64::from_be_bytes(b.try_into().map_err(|_| "a double of the wrong size")?)),            8 => json!(f32::from_be_bytes(b.try_into().map_err(|_| "a float of the wrong size")?) as f64),            9 => json!(i32::from_be_bytes(b.try_into().map_err(|_| "an int of the wrong size")?)),            0xb => json!(stamp(i64::from_be_bytes(b.try_into().map_err(|_| "a timestamp of the wrong size")?))),            0xc | 0xf => json!(uuid_text(b)),            0xe => json!(varint(b)),            0x10 => match b.len() {                4 => json!(IpAddr::from(<[u8; 4]>::try_from(b).unwrap_or_default()).to_string()),                16 => json!(IpAddr::from(<[u8; 16]>::try_from(b).unwrap_or_default()).to_string()),                _ => return Err("an inet of the wrong size".into()),            },            0x11 => json!(civil_iso(u32::from_be_bytes(b.try_into().map_err(|_| "a date of the wrong size")?) as i64 - (1i64 << 31))),            0x12 => {                let n = i64::from_be_bytes(b.try_into().map_err(|_| "a time of the wrong size")?);                json!(format!("{:02}:{:02}:{:02}.{:09}", n / 3_600_000_000_000, n / 60_000_000_000 % 60, n / 1_000_000_000 % 60, n % 1_000_000_000))            }            0x13 => json!(i16::from_be_bytes(b.try_into().map_err(|_| "a smallint of the wrong size")?)),            0x14 => json!(*b.first().ok_or("an empty tinyint")? as i8),            other => return Err(format!("CQL type {} is not known", other)),        },        Ty::Custom(_) => json!(format!("0x{}", hex::encode(b))),        Ty::List(e) | Ty::Set(e) => {            let mut r = Rd(b);            let n = r.i32()?;            Value::Array((0..n).map(|_| decode(e, r.bytes()?)).collect::<Result<_, _>>()?)        }        Ty::Map(k, v) => {            let mut r = Rd(b);            let n = r.i32()?;            let mut m = Map::new();            for _ in 0..n {                let key = decode(k, r.bytes()?)?;                m.insert(key.as_str().map(str::to_string).unwrap_or_else(|| key.to_string()), decode(v, r.bytes()?)?);            }            Value::Object(m)        }        Ty::Tuple(l) => {            let mut r = Rd(b);            Value::Array(l.iter().map(|e| decode(e, if r.0.is_empty() { None } else { r.bytes()? })).collect::<Result<_, _>>()?)        }        Ty::Udt(f) => {            let mut r = Rd(b);            let mut m = Map::new();            for (name, e) in f {                m.insert(name.clone(), decode(e, if r.0.is_empty() { None } else { r.bytes()? })?);            }            Value::Object(m)        }    })}fn num_i64(v: &Value, what: &str) -> Result<i64, String> {    v.as_i64().or(v.as_str().and_then(|s| s.trim().parse().ok())).ok_or(format!("{} is not {}", v, what))}fn num_f64(v: &Value, what: &str) -> Result<f64, String> {    v.as_f64().or(v.as_str().and_then(|s| s.trim().parse().ok())).ok_or(format!("{} is not {}", v, what))}fn uuid_bytes(s: &str) -> Result<Vec<u8>, String> {    let b = hex::decode(s.replace('-', "")).map_err(|_| format!("`{}` is not a uuid", s))?;    if b.len() != 16 {        return Err(format!("`{}` is not a uuid", s));    }    Ok(b)}fn put_bytes(out: &mut Vec<u8>, v: Option<&[u8]>) {    match v {        Some(b) => {            out.extend((b.len() as i32).to_be_bytes());            out.extend(b);        }        None => out.extend((-1i32).to_be_bytes()),    }}fn parse_stamp(s: &str) -> Result<i64, String> {    let days = days_of(s).ok_or(format!("`{}` is not a time", s))?;    let t = s.get(11..).unwrap_or("");    let n = |a: usize, b: usize| t.get(a..b).and_then(|x| x.parse::<i64>().ok()).unwrap_or(0);    let ms = t.get(9..12).and_then(|x| x.parse::<i64>().ok()).unwrap_or(0);    Ok(days * 86_400_000 + n(0, 2) * 3_600_000 + n(3, 5) * 60_000 + n(6, 8) * 1000 + ms)}/// A JSON value as the column's type wants it.fn encode(t: &Ty, v: &Value) -> Result<Option<Vec<u8>>, String> {    if v.is_null() {        return Ok(None);    }    Ok(Some(match t {        Ty::Plain(x) => match x {            1 | 0xd => v.as_str().map(str::to_string).unwrap_or_else(|| v.to_string()).into_bytes(),            2 | 5 => num_i64(v, "a bigint")?.to_be_bytes().to_vec(),            3 => hex::decode(v.as_str().ok_or(format!("{} is not hex text", v))?.trim_start_matches("0x")).map_err(|_| format!("{} is not hex", v))?,            4 => vec![v.as_bool().ok_or(format!("{} is not a boolean", v))? as u8],            6 => {                let text = v.as_str().map(str::to_string).unwrap_or_else(|| v.to_string());                let (a, b) = text.split_once('.').unwrap_or((&text, ""));                let mut out = (b.len() as i32).to_be_bytes().to_vec();                out.extend(varint_bytes(&format!("{}{}", a, b))?);                out            }            7 => num_f64(v, "a double")?.to_be_bytes().to_vec(),            8 => (num_f64(v, "a float")? as f32).to_be_bytes().to_vec(),            9 => i32::try_from(num_i64(v, "an int")?).map_err(|_| format!("{} is past an int", v))?.to_be_bytes().to_vec(),            0xb => match v {                Value::String(s) => parse_stamp(s)?,                _ => num_i64(v, "a timestamp")?,            }.to_be_bytes().to_vec(),            0xc | 0xf => uuid_bytes(v.as_str().ok_or(format!("{} is not a uuid", v))?)?,            0xe => varint_bytes(&v.as_str().map(str::to_string).unwrap_or_else(|| v.to_string()))?,            0x10 => match v.as_str().and_then(|s| s.parse::<IpAddr>().ok()) {                Some(IpAddr::V4(a)) => a.octets().to_vec(),                Some(IpAddr::V6(a)) => a.octets().to_vec(),                None => return Err(format!("{} is not an address", v)),            },            0x11 => (days_of(v.as_str().ok_or(format!("{} is not a date", v))?).ok_or(format!("{} is not a date", v))? + (1i64 << 31)).to_be_bytes()[4..].to_vec(),            0x13 => i16::try_from(num_i64(v, "a smallint")?).map_err(|_| format!("{} is past a smallint", v))?.to_be_bytes().to_vec(),            0x14 => vec![i8::try_from(num_i64(v, "a tinyint")?).map_err(|_| format!("{} is past a tinyint", v))? as u8],            other => return Err(format!("binding CQL type {} is not supported", other)),        },        Ty::List(e) | Ty::Set(e) => {            let a = v.as_array().ok_or(format!("{} is not a list", v))?;            let mut out = (a.len() as i32).to_be_bytes().to_vec();            for x in a {                put_bytes(&mut out, encode(e, x)?.as_deref());            }            out        }        Ty::Map(k, e) => {            let m = v.as_object().ok_or(format!("{} is not a map", v))?;            let mut out = (m.len() as i32).to_be_bytes().to_vec();            for (key, x) in m {                put_bytes(&mut out, encode(k, &json!(key))?.as_deref());                put_bytes(&mut out, encode(e, x)?.as_deref());            }            out        }        Ty::Tuple(l) => {            let a = v.as_array().ok_or(format!("{} is not a list", v))?;            let mut out = vec![];            for (e, x) in l.iter().zip(a) {                put_bytes(&mut out, encode(e, x)?.as_deref());            }            out        }        Ty::Udt(f) => {            let mut out = vec![];            for (name, e) in f {                put_bytes(&mut out, encode(e, v.get(name).unwrap_or(&Value::Null))?.as_deref());            }            out        }        Ty::Custom(n) => return Err(format!("binding the custom type {} is not supported", n)),    }))}/// A plain statement's values have no column types to go by, so they go by their own.fn encode_guess(v: &Value) -> Result<Option<Vec<u8>>, String> {    Ok(match v {        Value::Null => None,        Value::Bool(b) => Some(vec![*b as u8]),        Value::Number(n) if n.is_f64() => Some(n.as_f64().unwrap_or(0.0).to_be_bytes().to_vec()),        Value::Number(n) => Some(n.as_i64().unwrap_or(0).to_be_bytes().to_vec()),        Value::String(s) => Some(s.clone().into_bytes()),        other => return Err(format!("{} needs its type: use cassandra_execute, which prepares", other)),    })}fn consistency(o: &Value, k: &str, d: u16) -> Result<u16, String> {    let Some(v) = o.get(k).and_then(|v| v.as_str()) else { return Ok(d) };    Ok(match v.to_uppercase().as_str() {        "ANY" => 0, "ONE" => 1, "TWO" => 2, "THREE" => 3, "QUORUM" => 4, "ALL" => 5, "LOCAL_QUORUM" => 6, "EACH_QUORUM" => 7, "SERIAL" => 8, "LOCAL_SERIAL" => 9, "LOCAL_ONE" => 10,        other => return Err(format!("`{}` is not a consistency level", other)),    })}// ---- The connection ----struct Prepared {    id: Vec<u8>,    params: Vec<Ty>,    names: Vec<String>,}struct Conn {    s: TcpStream,    stream: i16,    prepared: HashMap<String, Arc<Prepared>>,    consistency: u16,    page_size: i32,    keyspace: Option<String>,    warnings: Vec<String>,}type Slot = Arc<Mutex<Option<Conn>>>;static CONNS: Mutex<Option<HashMap<String, (Value, Slot)>>> = Mutex::new(None);fn io(e: std::io::Error) -> String {    format!("the connection failed: {}", e)}fn error_of(body: &[u8]) -> String {    let mut r = Rd(body);    let Ok(code) = r.i32() else { return "a cut-off error".into() };    let msg = r.string().unwrap_or_default();    let kind = match code {        0x0000 => "server error", 0x000A => "protocol error", 0x0100 => "bad credentials", 0x1000 => "unavailable", 0x1001 => "overloaded", 0x1002 => "bootstrapping", 0x1003 => "truncate error",        0x1100 => "write timeout", 0x1200 => "read timeout", 0x1300 => "read failure", 0x1400 => "function failure", 0x1500 => "write failure", 0x2000 => "syntax error", 0x2100 => "unauthorized",        0x2200 => "invalid", 0x2300 => "config error", 0x2400 => "already exists", 0x2500 => "unprepared", _ => "error",    };    let more = match code {        0x1000 => format!(" ({} of {} replicas alive)", { let mut q = Rd(r.0); q.u16().ok(); q.i32().ok(); q.i32().unwrap_or(0) }, { let mut q = Rd(r.0); q.u16().ok(); q.i32().unwrap_or(0) }),        _ => String::new(),    };    format!("Cassandra {}: {}{}", kind, msg, more)}impl Conn {    fn send(&mut self, op: u8, body: &[u8]) -> Result<(), String> {        self.stream = if self.stream >= 32000 { 1 } else { self.stream + 1 };        let mut f = vec![0x04, 0];        f.extend(self.stream.to_be_bytes());        f.push(op);        f.extend((body.len() as u32).to_be_bytes());        f.extend(body);        self.s.write_all(&f).map_err(io)    }    fn recv(&mut self) -> Result<(u8, Vec<u8>), String> {        loop {            let mut h = [0u8; 9];            self.s.read_exact(&mut h).map_err(io)?;            if h[0] != 0x84 {                return Err(format!("a frame of protocol version {}: this speaks v4", h[0] & 0x7f));            }            let len = u32::from_be_bytes([h[5], h[6], h[7], h[8]]) as usize;            if len > 256 << 20 {                return Err("a frame over 256 MB".into());            }            let mut b = vec![0; len];            self.s.read_exact(&mut b).map_err(io)?;            let (flags, stream, op) = (h[1], i16::from_be_bytes([h[2], h[3]]), h[4]);            if stream != self.stream {                continue; // An event, or an old answer: this client sends one at a time.            }            let mut r = Rd(&b);            if flags & 2 != 0 {                r.take(16)?;            }            if flags & 8 != 0 {                self.warnings = r.string_list()?;            }            if flags & 4 != 0 {                for _ in 0..r.u16()? {                    r.string()?;                    r.bytes()?;                }            }            return Ok((op, r.0.to_vec()));        }    }    fn ask(&mut self, op: u8, body: &[u8]) -> Result<(u8, Vec<u8>), String> {        self.send(op, body)?;        self.recv()    }}fn str_map(pairs: &[(&str, &str)]) -> Vec<u8> {    let mut b = (pairs.len() as u16).to_be_bytes().to_vec();    for (k, v) in pairs {        b.extend((k.len() as u16).to_be_bytes());        b.extend(k.as_bytes());        b.extend((v.len() as u16).to_be_bytes());        b.extend(v.as_bytes());    }    b}fn open(o: &Value) -> Result<Conn, String> {    let host = o.get("host").and_then(|h| h.as_str()).map(str::to_string).or_else(|| std::env::var("CASSANDRA_HOST").ok()).unwrap_or("127.0.0.1:9042".into());    let addr = if host.contains(':') { host.clone() } else { format!("{}:9042", host) };    let t = Duration::from_millis(o.get("timeout_ms").and_then(|t| t.as_u64()).unwrap_or(15_000));    let list: Vec<_> = addr.to_socket_addrs().map_err(|e| format!("{}: {}", addr, e))?.collect();    let sa = list.first().ok_or(format!("{} has no address", addr))?;    let s = TcpStream::connect_timeout(sa, t).map_err(|e| format!("cannot reach Cassandra at {}: {}", addr, e))?;    s.set_read_timeout(Some(t)).map_err(io)?;    s.set_write_timeout(Some(t)).map_err(io)?;    s.set_nodelay(true).map_err(io)?;    let mut c = Conn { s, stream: 0, prepared: HashMap::new(), consistency: consistency(o, "consistency", 6)?, page_size: o.get("page_size").and_then(|p| p.as_i64()).unwrap_or(5000) as i32, keyspace: None, warnings: vec![] };    let (op, body) = c.ask(STARTUP, &str_map(&[("CQL_VERSION", "3.0.0")]))?;    match op {        READY => {}        AUTHENTICATE => {            let user = std::env::var("CASSANDRA_USER").ok().filter(|v| !v.is_empty()).ok_or("the server wants a password: set CASSANDRA_USER and CASSANDRA_PASSWORD")?;            let pass = std::env::var("CASSANDRA_PASSWORD").unwrap_or_default();            let mut token = vec![0];            token.extend(user.as_bytes());            token.push(0);            token.extend(pass.as_bytes());            let mut b = vec![];            put_bytes(&mut b, Some(&token));            let mut step = c.ask(AUTH_RESPONSE, &b)?;            // A challenge wants the same SASL PLAIN token again; PasswordAuthenticator has none.            let mut rounds = 0;            while step.0 == AUTH_CHALLENGE && rounds < 4 {                step = c.ask(AUTH_RESPONSE, &b)?;                rounds += 1;            }            match step.0 {                AUTH_SUCCESS => {}                0 => return Err(error_of(&step.1)),                x => return Err(format!("opcode {} during authentication", x)),            }        }        0 => return Err(error_of(&body)),        x => return Err(format!("opcode {} in answer to STARTUP", x)),    }    if let Some(k) = o.get("keyspace").and_then(|k| k.as_str()) {        use_keyspace(&mut c, k)?;    }    Ok(c)}fn long_str(s: &str) -> Vec<u8> {    let mut b = (s.len() as u32).to_be_bytes().to_vec();    b.extend(s.as_bytes());    b}fn use_keyspace(c: &mut Conn, k: &str) -> Result<(), String> {    if !k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_') || k.is_empty() {        return Err(format!("`{}` is not a keyspace name", k));    }    let mut b = long_str(&format!("USE {}", k));    b.extend(1u16.to_be_bytes());    b.push(0);    let (op, body) = c.ask(QUERY, &b)?;    if op == 0 {        return Err(error_of(&body));    }    c.keyspace = Some(k.to_string());    Ok(())}fn slot(name: &str) -> Result<(Value, Slot), String> {    let g = CONNS.lock().map_err(|e| e.to_string())?;    g.as_ref().and_then(|m| m.get(name)).cloned().ok_or(format!("no connection called `{}`: call cassandra_connect first", name))}/// Runs `f` on the connection, opening it again once if it had died.fn with<T>(name: &str, mut f: impl FnMut(&mut Conn) -> Result<T, String>) -> Result<T, String> {    let (cfg, s) = slot(name)?;    let mut g = s.lock().map_err(|e| e.to_string())?;    if g.is_none() {        *g = Some(open(&cfg)?);    }    match f(g.as_mut().unwrap()) {        Err(e) if e.starts_with("the connection failed") => {            *g = Some(open(&cfg)?);            f(g.as_mut().unwrap())        }        r => r,    }}/// Connects and logs in: `options` is { host (else CASSANDRA_HOST, "host:port"), keyspace, consistency (LOCAL_QUORUM), page_size (5000), timeout_ms }. The password comes from CASSANDRA_USER and CASSANDRA_PASSWORD. Gives { connected, supported }.pub fn cassandra_connect(name: &str, options: serde_json::Value) -> Result<serde_json::Value, String> {    let mut c = open(&options)?;    let (op, body) = c.ask(OPTIONS, &[])?;    let mut supported = Map::new();    if op == SUPPORTED {        let mut r = Rd(&body);        for _ in 0..r.u16()? {            let k = r.string()?;            supported.insert(k, json!(r.string_list()?));        }    }    let mut g = CONNS.lock().map_err(|e| e.to_string())?;    g.get_or_insert_with(HashMap::new).insert(name.to_string(), (options, Arc::new(Mutex::new(Some(c)))));    Ok(json!({ "connected": true, "supported": supported }))}/// Closes a connection.pub fn cassandra_close(name: &str) -> Result<bool, String> {    let mut g = CONNS.lock().map_err(|e| e.to_string())?;    Ok(g.as_mut().and_then(|m| m.remove(name)).is_some())}fn read_columns(r: &mut Rd, n: i32, global: bool) -> Result<Vec<(String, Ty)>, String> {    if global {        r.string()?;        r.string()?;    }    (0..n).map(|_| {        if !global {            r.string()?;            r.string()?;        }        let name = r.string()?;        Ok((name, read_type(r)?))    }).collect()}fn read_result(body: &[u8], cached: Option<&Prepared>) -> Result<Value, String> {    let _ = cached;    let mut r = Rd(body);    Ok(match r.i32()? {        1 => json!({ "rows": [], "columns": [], "paging_state": null, "has_more": false }),        2 => {            let flags = r.i32()?;            let n = r.i32()?;            let paging = if flags & 2 != 0 { r.bytes()?.map(hex::encode) } else { None };            let cols = read_columns(&mut r, n, flags & 1 != 0)?;            let count = r.i32()?;            let mut rows = vec![];            for _ in 0..count {                let mut m = Map::new();                for (name, t) in &cols {                    m.insert(name.clone(), decode(t, r.bytes()?)?);                }                rows.push(Value::Object(m));            }            let c: Vec<Value> = cols.iter().map(|(n, t)| json!({ "name": n, "type": type_name(t) })).collect();            json!({ "rows": rows, "columns": c, "paging_state": paging, "has_more": flags & 2 != 0 })        }        3 => json!({ "keyspace": r.string()?, "rows": [], "columns": [], "paging_state": null, "has_more": false }),        5 => json!({ "schema_change": r.string()?, "target": r.string()?, "rows": [], "columns": [], "paging_state": null, "has_more": false }),        k => return Err(format!("result kind {} is not expected here", k)),    })}/// [consistency][flags][values][page size][paging state][serial]...fn params(c: &Conn, o: &Value, values: Vec<Option<Vec<u8>>>) -> Result<Vec<u8>, String> {    let mut b = consistency(o, "consistency", c.consistency)?.to_be_bytes().to_vec();    let page = o.get("page_size").and_then(|p| p.as_i64()).unwrap_or(c.page_size as i64) as i32;    let paging = o.get("paging_state").and_then(|p| p.as_str()).filter(|p| !p.is_empty()).map(hex::decode).transpose().map_err(|_| "paging_state is not hex".to_string())?;    let serial = o.get("serial_consistency").is_some();    let ts = o.get("timestamp").and_then(|t| t.as_i64());    let mut f = 0u8;    if !values.is_empty() {        f |= 1;    }    if page > 0 {        f |= 4;    }    if paging.is_some() {        f |= 8;    }    if serial {        f |= 0x10;    }    if ts.is_some() {        f |= 0x20;    }    b.push(f);    if !values.is_empty() {        b.extend((values.len() as u16).to_be_bytes());        for v in &values {            put_bytes(&mut b, v.as_deref());        }    }    if page > 0 {        b.extend(page.to_be_bytes());    }    if let Some(p) = &paging {        put_bytes(&mut b, Some(p));    }    if serial {        b.extend(consistency(o, "serial_consistency", 8)?.to_be_bytes());    }    if let Some(t) = ts {        b.extend(t.to_be_bytes());    }    Ok(b)}fn plain(c: &mut Conn, cql: &str, values: &Value, o: &Value) -> Result<Value, String> {    let list = values.as_array().map(|a| a.iter().map(encode_guess).collect::<Result<Vec<_>, _>>()).transpose()?.unwrap_or_default();    let mut b = long_str(cql);    b.extend(params(c, o, list)?);    let (op, body) = c.ask(QUERY, &b)?;    match op {        RESULT => read_result(&body, None),        0 => Err(error_of(&body)),        x => Err(format!("opcode {} in answer to QUERY", x)),    }}fn prepare(c: &mut Conn, cql: &str) -> Result<Arc<Prepared>, String> {    if let Some(p) = c.prepared.get(cql) {        return Ok(p.clone());    }    let fresh = prepare_new(c, cql)?;    Ok(fresh)}fn prepare_new(c: &mut Conn, cql: &str) -> Result<Arc<Prepared>, String> {    let (op, body) = c.ask(PREPARE, &long_str(cql))?;    if op == 0 {        return Err(error_of(&body));    }    if op != RESULT {        return Err(format!("opcode {} in answer to PREPARE", op));    }    let mut r = Rd(&body);    if r.i32()? != 4 {        return Err("PREPARE gave no prepared statement".into());    }    let id = r.short_bytes()?.to_vec();    let flags = r.i32()?;    let n = r.i32()?;    let pk = r.i32()?;    for _ in 0..pk {        r.u16()?;    }    let cols = read_columns(&mut r, n, flags & 1 != 0)?;    let p = Arc::new(Prepared { id, names: cols.iter().map(|c| c.0.clone()).collect(), params: cols.into_iter().map(|c| c.1).collect() });    c.prepared.insert(cql.to_string(), p.clone());    Ok(p)}fn bind(p: &Prepared, values: &Value) -> Result<Vec<Option<Vec<u8>>>, String> {    let list: Vec<Value> = match values {        Value::Array(a) => a.clone(),        Value::Object(m) => p.names.iter().map(|n| m.get(n).cloned().unwrap_or(Value::Null)).collect(),        Value::Null => vec![],        other => return Err(format!("values are a list or a map, not {}", other)),    };    if list.len() != p.params.len() {        return Err(format!("the statement has {} markers, and {} values came", p.params.len(), list.len()));    }    list.iter().zip(&p.params).zip(&p.names).map(|((v, t), n)| encode(t, v).map_err(|e| format!("{}: {}", n, e))).collect()}fn exec(c: &mut Conn, cql: &str, values: &Value, o: &Value) -> Result<Value, String> {    let mut p = prepare(c, cql)?;    for attempt in 0..2 {        let mut b = (p.id.len() as u16).to_be_bytes().to_vec();        b.extend(&p.id);        b.extend(params(c, o, bind(&p, values)?)?);        let (op, body) = c.ask(EXECUTE, &b)?;        match op {            RESULT => return read_result(&body, Some(&p)),            0 => {                let code = Rd(&body).i32().unwrap_or(0);                // The server evicted it, or restarted: prepare again, once.                if code == 0x2500 && attempt == 0 {                    c.prepared.remove(cql);                    p = prepare_new(c, cql)?;                    continue;                }                return Err(error_of(&body));            }            x => return Err(format!("opcode {} in answer to EXECUTE", x)),        }    }    Err("the statement stayed unprepared".into())}/// Runs a statement with its text: `values` a list or null, sent by their own JSON types (int as bigint, float as double, text), which only fits such columns. Options: consistency, page_size, paging_state, serial_consistency, timestamp. Gives { rows, columns: [{ name, type }], paging_state, has_more }: one page.pub fn cassandra_query(name: &str, cql: &str, values: serde_json::Value, options: serde_json::Value) -> Result<serde_json::Value, String> {    with(name, |c| plain(c, cql, &values, &options))}/// Runs a statement prepared once per connection, its values encoded by the column types the server gave: a list, or a map by marker name. An Unprepared answer prepares it again. Options and answer as for `cassandra_query`.pub fn cassandra_execute(name: &str, cql: &str, values: serde_json::Value, options: serde_json::Value) -> Result<serde_json::Value, String> {    with(name, |c| exec(c, cql, &values, &options))}/// Every page of a prepared statement, following paging_state, up to the option max_rows. Gives { rows, pages }.pub fn cassandra_all(name: &str, cql: &str, values: serde_json::Value, options: serde_json::Value) -> Result<serde_json::Value, String> {    let most = options.get("max_rows").and_then(|m| m.as_u64()).unwrap_or(u64::MAX) as usize;    let mut o = if options.is_object() { options } else { json!({}) };    let (mut rows, mut pages) = (vec![], 0);    loop {        let r = cassandra_execute(name, cql, values.clone(), o.clone())?;        pages += 1;        rows.extend(r["rows"].as_array().cloned().unwrap_or_default());        if r["has_more"] != true || rows.len() >= most {            rows.truncate(most);            return Ok(json!({ "rows": rows, "pages": pages }));        }        o["paging_state"] = r["paging_state"].clone();    }}/// Runs statements as one batch: `statements` is [{ cql, values }]; each is prepared. Options: kind (logged, unlogged, counter; logged by default), consistency.pub fn cassandra_batch(name: &str, statements: Vec<serde_json::Value>, options: serde_json::Value) -> Result<bool, String> {    let kind: u8 = match options.get("kind").and_then(|k| k.as_str()).unwrap_or("logged") {        "logged" => 0,        "unlogged" => 1,        "counter" => 2,        other => return Err(format!("`{}` is not a batch kind: logged, unlogged or counter", other)),    };    with(name, |c| {        let mut b = vec![kind];        b.extend((statements.len() as u16).to_be_bytes());        for st in &statements {            let cql = st["cql"].as_str().ok_or("a batch statement needs `cql`")?;            let p = prepare(c, cql)?;            b.push(1);            b.extend((p.id.len() as u16).to_be_bytes());            b.extend(&p.id);            let vals = bind(&p, st.get("values").unwrap_or(&Value::Null))?;            b.extend((vals.len() as u16).to_be_bytes());            for v in &vals {                put_bytes(&mut b, v.as_deref());            }        }        b.extend(consistency(&options, "consistency", c.consistency)?.to_be_bytes());        b.push(0);        let (op, body) = c.ask(BATCH, &b)?;        match op {            RESULT => Ok(true),            0 => Err(error_of(&body)),            x => Err(format!("opcode {} in answer to BATCH", x)),        }    })}/// Switches a connection to a keyspace.pub fn cassandra_use(name: &str, keyspace: &str) -> Result<bool, String> {    with(name, |c| use_keyspace(c, keyspace))?;    Ok(true)}/// Warnings the server attached to the last answer, like a large batch.pub fn cassandra_warnings(name: &str) -> Result<Vec<String>, String> {    with(name, |c| Ok(c.warnings.clone()))}