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

reddb_client/redwire/
mod.rs
