reddb_client/redwire/
mod.rs

1//! RedWire client.
2//!
3//! Uses the canonical RedWire contracts from `reddb-wire` so the
4//! client does not duplicate the engine-facing frame and payload
5//! definitions. The framing is a stable wire contract defined by
6//! ADR 0001 (`.red/adr/0001-redwire-tcp-protocol.md`).
7//!
8//! Public surface:
9//!   - [`RedWireClient::connect`][]: TCP + handshake + auth
10//!   - [`RedWireClient::query`][]: SQL → server result
11//!   - [`RedWireClient::ping`][]: keepalive
12//!   - [`RedWireClient::close`][]: clean shutdown via Bye
13
14mod handshake;
15mod io;
16#[cfg(feature = "redwire")]
17pub mod scram;
18#[cfg(feature = "redwire-tls")]
19mod tls;
20
21pub use reddb_wire::redwire::{Flags, Frame, FrameError, MessageKind};
22#[cfg(feature = "redwire-tls")]
23pub use tls::TlsConfig;
24
25use std::io as std_io;
26use std::pin::Pin;
27
28use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
29use tokio::net::TcpStream;
30
31use reddb_wire::query_with_params::{
32    encode_query_with_params, ParamValue as RedWireParamValue, FEATURE_PARAMS,
33};
34
35/// Boxed read+write trait object. Lets `RedWireClient` carry
36/// either a plain `TcpStream` or a `tokio_rustls::TlsStream`
37/// without leaking the type up.
38pub(crate) type Stream = Pin<Box<dyn AsyncReadWrite + Send + Unpin>>;
39
40pub(crate) trait AsyncReadWrite: AsyncRead + AsyncWrite {}
41impl<T: AsyncRead + AsyncWrite + ?Sized> AsyncReadWrite for T {}
42
43use crate::error::{ClientError, ErrorCode, Result};
44use crate::types::{BulkInsertResult, QueryResult};
45
46use handshake::HandshakeOutcome;
47use reddb_wire::legacy::WireValue;
48use reddb_wire::redwire::{
49    build_bulk_insert_binary_frame, build_bulk_insert_frame, build_bye_frame, build_delete_frame,
50    build_get_frame, build_ping_frame, build_query_frame, build_query_with_params_frame,
51    decode_bulk_ok_count_payload, decode_bulk_ok_payload, decode_delete_ok_affected,
52    decode_get_result_payload, decode_query_result_payload, decode_text_payload,
53    encode_bulk_binary_payload, encode_bulk_insert_payload, encode_insert_payload,
54    encode_key_payload, expect_bulk_ok_or_error, expect_delete_ok_or_error, expect_pong_reply,
55    expect_result_or_error, supported_client_preface, BuildError, OperationReplyError,
56};
57
58/// Authentication credentials for the RedWire handshake.
59#[derive(Debug, Clone)]
60pub enum Auth {
61    /// Server is configured with `auth.enabled = false`.
62    Anonymous,
63    /// Bearer token (login-derived session token or API key).
64    Bearer(String),
65}
66
67/// Typed value for the binary bulk-insert fast path. Encoding delegates
68/// to `reddb-wire::legacy`, which owns the byte-level value tags.
69#[derive(Debug, Clone)]
70pub enum BinaryValue {
71    I64(i64),
72    F64(f64),
73    Text(String),
74    Bool(bool),
75    Null,
76}
77
78impl From<&BinaryValue> for WireValue {
79    fn from(value: &BinaryValue) -> Self {
80        match value {
81            BinaryValue::I64(n) => WireValue::I64(*n),
82            BinaryValue::F64(n) => WireValue::F64(*n),
83            BinaryValue::Text(value) => WireValue::Text(value.clone()),
84            BinaryValue::Bool(value) => WireValue::Bool(*value),
85            BinaryValue::Null => WireValue::Null,
86        }
87    }
88}
89
90/// Configuration for `RedWireClient::connect`.
91pub struct ConnectOptions {
92    pub host: String,
93    pub port: u16,
94    pub auth: Auth,
95    pub client_name: Option<String>,
96    #[cfg(feature = "redwire-tls")]
97    pub tls: Option<TlsConfig>,
98}
99
100impl std::fmt::Debug for ConnectOptions {
101    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102        let mut s = f.debug_struct("ConnectOptions");
103        s.field("host", &self.host)
104            .field("port", &self.port)
105            .field("auth", &self.auth)
106            .field("client_name", &self.client_name);
107        #[cfg(feature = "redwire-tls")]
108        s.field("tls", &self.tls.is_some());
109        s.finish()
110    }
111}
112
113impl Clone for ConnectOptions {
114    fn clone(&self) -> Self {
115        Self {
116            host: self.host.clone(),
117            port: self.port,
118            auth: self.auth.clone(),
119            client_name: self.client_name.clone(),
120            #[cfg(feature = "redwire-tls")]
121            tls: self.tls.clone(),
122        }
123    }
124}
125
126impl ConnectOptions {
127    pub fn new(host: impl Into<String>, port: u16) -> Self {
128        Self {
129            host: host.into(),
130            port,
131            auth: Auth::Anonymous,
132            client_name: Some(format!("reddb-rs/{}", env!("CARGO_PKG_VERSION"))),
133            #[cfg(feature = "redwire-tls")]
134            tls: None,
135        }
136    }
137
138    pub fn with_auth(mut self, auth: Auth) -> Self {
139        self.auth = auth;
140        self
141    }
142
143    pub fn with_client_name(mut self, name: impl Into<String>) -> Self {
144        self.client_name = Some(name.into());
145        self
146    }
147
148    /// Wrap the TCP socket in TLS using the supplied config.
149    #[cfg(feature = "redwire-tls")]
150    pub fn with_tls(mut self, tls: TlsConfig) -> Self {
151        self.tls = Some(tls);
152        self
153    }
154}
155
156pub struct RedWireClient {
157    stream: Stream,
158    next_correlation_id: u64,
159    #[allow(dead_code)]
160    session_id: String,
161    #[allow(dead_code)]
162    server_features: u32,
163}
164
165impl std::fmt::Debug for RedWireClient {
166    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
167        f.debug_struct("RedWireClient")
168            .field("session_id", &self.session_id)
169            .field("server_features", &self.server_features)
170            .finish()
171    }
172}
173
174impl RedWireClient {
175    pub async fn connect(opts: ConnectOptions) -> Result<Self> {
176        let addr = format!("{}:{}", opts.host, opts.port);
177        let tcp = TcpStream::connect(&addr)
178            .await
179            .map_err(|e| ClientError::new(ErrorCode::Network, format!("{addr}: {e}")))?;
180
181        let mut stream: Stream = match () {
182            #[cfg(feature = "redwire-tls")]
183            _ if opts.tls.is_some() => {
184                let tls_cfg = opts.tls.as_ref().unwrap();
185                let tls_stream = tls::wrap_client(tcp, &opts.host, tls_cfg).await?;
186                Box::pin(tls_stream)
187            }
188            _ => Box::pin(tcp),
189        };
190
191        // Discriminator + minor-version byte.
192        stream
193            .write_all(&supported_client_preface())
194            .await
195            .map_err(io_err)?;
196
197        let outcome = handshake::run(&mut stream, &opts).await?;
198        match outcome {
199            HandshakeOutcome::Authenticated {
200                session_id,
201                server_features,
202            } => Ok(Self {
203                stream,
204                next_correlation_id: 1,
205                session_id,
206                server_features,
207            }),
208            HandshakeOutcome::Refused(reason) => Err(ClientError::new(
209                ErrorCode::AuthRefused,
210                format!("redwire auth refused: {reason}"),
211            )),
212        }
213    }
214
215    pub async fn query(&mut self, sql: &str) -> Result<QueryResult> {
216        let raw = self.query_raw(sql).await?;
217        let value = decode_query_result_payload(raw.as_bytes())
218            .map_err(|e| ClientError::new(ErrorCode::Protocol, format!("decode result: {e}")))?;
219        Ok(QueryResult::from_envelope(value))
220    }
221
222    /// Send a query and return the raw `Result` payload string. This is
223    /// used by the legacy `red_client` compatibility shim so the CLI
224    /// output stays byte-for-byte aligned with its old RedWire path.
225    pub async fn query_raw(&mut self, sql: &str) -> Result<String> {
226        let corr = self.next_corr();
227        let req = build_query_frame(corr, sql).map_err(frame_build_err)?;
228        io::write_frame(&mut self.stream, &req).await?;
229        let resp = self.read_frame().await?;
230        let payload = expect_result_or_error(resp.kind, &resp.payload).map_err(reply_err)?;
231        Ok(decode_text_payload(payload))
232    }
233
234    pub async fn query_with(
235        &mut self,
236        sql: &str,
237        params: &[crate::params::Value],
238    ) -> Result<QueryResult> {
239        if params.is_empty() {
240            return self.query(sql).await;
241        }
242        if self.server_features & FEATURE_PARAMS == 0 {
243            return Err(ClientError::new(
244                ErrorCode::ParamsUnsupported,
245                "server did not advertise RedWire parameter support",
246            ));
247        }
248
249        let wire_params = params.iter().map(param_to_redwire).collect::<Vec<_>>();
250        let payload = encode_query_with_params(sql, &wire_params)
251            .map_err(|e| ClientError::new(ErrorCode::Protocol, format!("encode params: {e}")))?;
252        let corr = self.next_corr();
253        let req = build_query_with_params_frame(corr, payload).map_err(frame_build_err)?;
254        io::write_frame(&mut self.stream, &req).await?;
255        let resp = self.read_frame().await?;
256        let payload = expect_result_or_error(resp.kind, &resp.payload).map_err(reply_err)?;
257        let value = decode_query_result_payload(payload)
258            .map_err(|e| ClientError::new(ErrorCode::Protocol, format!("decode result: {e}")))?;
259        Ok(QueryResult::from_envelope(value))
260    }
261
262    /// Insert a single row. `payload` is a JSON object with column
263    /// → value pairs. Returns the engine's affected-rows count.
264    pub async fn insert(&mut self, collection: &str, payload: serde_json::Value) -> Result<u64> {
265        self.send_insert_frame(encode_insert_payload(collection, payload))
266            .await
267            .map(|result| result.affected)
268    }
269
270    /// Bulk insert. Each entry in `payloads` is a JSON object.
271    pub async fn bulk_insert(
272        &mut self,
273        collection: &str,
274        payloads: Vec<serde_json::Value>,
275    ) -> Result<BulkInsertResult> {
276        self.send_insert_frame(encode_bulk_insert_payload(collection, payloads))
277            .await
278    }
279
280    async fn send_insert_frame(&mut self, bytes: Vec<u8>) -> Result<BulkInsertResult> {
281        let corr = self.next_corr();
282        let req = build_bulk_insert_frame(corr, bytes).map_err(frame_build_err)?;
283        io::write_frame(&mut self.stream, &req).await?;
284        let resp = self.read_frame().await?;
285        let payload = expect_bulk_ok_or_error(resp.kind, &resp.payload).map_err(reply_err)?;
286        let payload = decode_bulk_ok_payload(payload)
287            .map_err(|e| ClientError::new(ErrorCode::Protocol, format!("decode bulk_ok: {e}")))?;
288        Ok(BulkInsertResult {
289            affected: payload.affected,
290            rids: payload.rids,
291            ids: payload.ids,
292        })
293    }
294
295    /// Fetch one row by primary id. Returns the JSON envelope the
296    /// server emits on a `Get` frame: `{ ok, found, ... }`.
297    pub async fn get(&mut self, collection: &str, id: &str) -> Result<serde_json::Value> {
298        let corr = self.next_corr();
299        let req =
300            build_get_frame(corr, encode_key_payload(collection, id)).map_err(frame_build_err)?;
301        io::write_frame(&mut self.stream, &req).await?;
302        let resp = self.read_frame().await?;
303        let payload = expect_result_or_error(resp.kind, &resp.payload).map_err(reply_err)?;
304        decode_get_result_payload(payload)
305            .map_err(|e| ClientError::new(ErrorCode::Protocol, format!("decode get: {e}")))
306    }
307
308    /// Delete by primary id. Returns the affected count.
309    pub async fn delete(&mut self, collection: &str, id: &str) -> Result<u64> {
310        let corr = self.next_corr();
311        let req = build_delete_frame(corr, encode_key_payload(collection, id))
312            .map_err(frame_build_err)?;
313        io::write_frame(&mut self.stream, &req).await?;
314        let resp = self.read_frame().await?;
315        let payload = expect_delete_ok_or_error(resp.kind, &resp.payload).map_err(reply_err)?;
316        decode_delete_ok_affected(payload)
317            .map_err(|e| ClientError::new(ErrorCode::Protocol, format!("decode delete_ok: {e}")))
318    }
319
320    /// Bulk-insert via the binary fast path. Same wire shape as
321    /// `MSG_BULK_INSERT_BINARY` (0x06): typed values, no JSON
322    /// encode/decode. Use this for hot inserts where the column
323    /// types are known up front.
324    ///
325    /// `columns`: column names in fixed order.
326    /// `rows`: each inner Vec must have one entry per column, in
327    /// column order. Mixed types are encoded by the value writer.
328    pub async fn bulk_insert_binary(
329        &mut self,
330        collection: &str,
331        columns: &[&str],
332        rows: &[Vec<BinaryValue>],
333    ) -> Result<u64> {
334        let wire_rows = rows
335            .iter()
336            .map(|row| row.iter().map(WireValue::from).collect::<Vec<_>>())
337            .collect::<Vec<_>>();
338        let payload = encode_bulk_binary_payload(collection, columns, &wire_rows)
339            .map_err(|e| ClientError::new(ErrorCode::Protocol, e.to_string()))?;
340
341        let corr = self.next_corr();
342        let req = build_bulk_insert_binary_frame(corr, payload).map_err(frame_build_err)?;
343        io::write_frame(&mut self.stream, &req).await?;
344        let resp = self.read_frame().await?;
345        let payload = expect_bulk_ok_or_error(resp.kind, &resp.payload).map_err(reply_err)?;
346        decode_bulk_ok_count_payload(payload)
347            .map_err(|e| ClientError::new(ErrorCode::Protocol, e.to_string()))
348    }
349
350    pub async fn ping(&mut self) -> Result<()> {
351        let corr = self.next_corr();
352        let req = build_ping_frame(corr).map_err(frame_build_err)?;
353        io::write_frame(&mut self.stream, &req).await?;
354        let resp = self.read_frame().await?;
355        expect_pong_reply(resp.kind).map_err(reply_err)
356    }
357
358    pub async fn close(mut self) -> Result<()> {
359        let corr = self.next_corr();
360        let bye = build_bye_frame(corr).map_err(frame_build_err)?;
361        let _ = io::write_frame(&mut self.stream, &bye).await;
362        Ok(())
363    }
364
365    fn next_corr(&mut self) -> u64 {
366        let c = self.next_correlation_id;
367        self.next_correlation_id = self.next_correlation_id.wrapping_add(1);
368        c
369    }
370
371    async fn read_frame(&mut self) -> Result<Frame> {
372        io::read_frame(&mut self.stream).await
373    }
374}
375
376fn param_to_redwire(value: &crate::params::Value) -> RedWireParamValue {
377    match value {
378        crate::params::Value::Null => RedWireParamValue::Null,
379        crate::params::Value::Bool(value) => RedWireParamValue::Bool(*value),
380        crate::params::Value::Int(value) => RedWireParamValue::Int(*value),
381        crate::params::Value::Float(value) => RedWireParamValue::Float(*value),
382        crate::params::Value::Text(value) => RedWireParamValue::Text(value.clone()),
383        crate::params::Value::Bytes(value) => RedWireParamValue::Bytes(value.clone()),
384        crate::params::Value::Vector(value) => RedWireParamValue::Vector(value.clone()),
385        crate::params::Value::Json(value) => {
386            RedWireParamValue::Json(value.to_json_string().into_bytes())
387        }
388        crate::params::Value::Timestamp(value) => RedWireParamValue::Timestamp(*value),
389        crate::params::Value::Uuid(value) => RedWireParamValue::Uuid(*value),
390    }
391}
392
393fn io_err(err: std_io::Error) -> ClientError {
394    ClientError::new(ErrorCode::Network, err.to_string())
395}
396
397fn frame_build_err(err: BuildError) -> ClientError {
398    ClientError::new(ErrorCode::Protocol, format!("build redwire frame: {err}"))
399}
400
401fn reply_err(err: OperationReplyError) -> ClientError {
402    match err {
403        OperationReplyError::Engine(message) => ClientError::new(ErrorCode::Engine, message),
404        OperationReplyError::UnexpectedKind { .. } => {
405            ClientError::new(ErrorCode::Protocol, err.to_string())
406        }
407    }
408}
409
410#[cfg(test)]
411mod tests {
412    use super::param_to_redwire;
413    use crate::{JsonValue, Value};
414    use reddb_wire::query_with_params::ParamValue as WireValue;
415
416    #[test]
417    fn param_to_redwire_preserves_all_wire_variants() {
418        let uuid = [0x11; 16];
419        let cases = vec![
420            (Value::Null, WireValue::Null),
421            (Value::Bool(true), WireValue::Bool(true)),
422            (Value::Int64(42), WireValue::Int(42)),
423            (Value::Float(1.5), WireValue::Float(1.5)),
424            (Value::Text("Ada".into()), WireValue::Text("Ada".into())),
425            (
426                Value::Bytes(vec![0xde, 0xad]),
427                WireValue::Bytes(vec![0xde, 0xad]),
428            ),
429            (
430                Value::Vector(vec![0.25, 0.5]),
431                WireValue::Vector(vec![0.25, 0.5]),
432            ),
433            (
434                Value::Json(JsonValue::object([("role", JsonValue::string("admin"))])),
435                WireValue::Json(br#"{"role":"admin"}"#.to_vec()),
436            ),
437            (
438                Value::Timestamp(1_700_000_000),
439                WireValue::Timestamp(1_700_000_000),
440            ),
441            (Value::Uuid(uuid), WireValue::Uuid(uuid)),
442        ];
443
444        for (input, expected) in cases {
445            assert_eq!(param_to_redwire(&input), expected);
446        }
447    }
448}
reddb_client/redwire/mod.rs

reddb_client/redwire/
mod.rs
