reifydb_client/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3#![cfg_attr(not(debug_assertions), deny(clippy::disallowed_methods))]
4#![cfg_attr(debug_assertions, warn(clippy::disallowed_methods))]
5#![allow(clippy::tabs_in_doc_comments)]
6
7#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
8#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
9#[cfg_attr(any(feature = "http", feature = "ws"), serde(rename_all = "lowercase"))]
10pub enum WireFormat {
11	#[default]
12	Frames,
13	Rbcf,
14}
15
16#[cfg(any(feature = "ws", feature = "grpc"))]
17mod changes;
18#[cfg(any(feature = "ws", feature = "grpc"))]
19pub mod client;
20#[cfg(all(feature = "dst", reifydb_single_threaded))]
21pub mod dst;
22#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
23pub mod error;
24#[cfg(feature = "grpc")]
25pub mod grpc;
26#[cfg(feature = "http")]
27pub mod http;
28#[cfg(any(feature = "ws", feature = "grpc"))]
29mod reconnect;
30#[cfg(any(feature = "http", feature = "ws"))]
31mod session;
32#[cfg(any(feature = "ws", feature = "grpc", all(feature = "dst", reifydb_single_threaded)))]
33pub mod subscription;
34#[cfg(feature = "ws")]
35mod utils;
36#[cfg(feature = "ws")]
37pub mod ws;
38
39#[cfg(any(feature = "http", feature = "ws"))]
40use std::collections::HashMap;
41#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
42use std::sync::Arc;
43
44#[cfg(all(feature = "dst", reifydb_single_threaded))]
45pub use dst::DstClient;
46#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
47pub use error::ClientError;
48#[cfg(feature = "grpc")]
49pub use grpc::{
50	BatchFramesEnvelope, BatchGrpcSubscription, BatchMemberHandle, BatchStreamEvent, GrpcChange, GrpcClient,
51	GrpcClientOptions, GrpcSubscription, RawChangePayload,
52};
53#[cfg(feature = "http")]
54pub use http::HttpClient;
55pub use reifydb_client_derive::FromFrame;
56pub use reifydb_value as value;
57#[cfg(any(feature = "ws", feature = "grpc"))]
58use reifydb_value::error::Error;
59pub use reifydb_value::{
60	params::Params,
61	value::{
62		Value,
63		frame::{
64			column::FrameColumn,
65			data::FrameColumnData,
66			extract::FrameError,
67			frame::Frame,
68			from_frame::FromFrameError,
69			row::{FrameRow, FrameRows},
70		},
71		iso::{IsoDate, IsoDateTime, IsoDuration, IsoTime},
72		ordered_f32::OrderedF32,
73		ordered_f64::OrderedF64,
74		try_from::{FromValueError, TryFromValue, TryFromValueCoerce},
75		value_type::ValueType,
76	},
77};
78#[cfg(any(feature = "http", feature = "ws"))]
79use serde::{Deserialize, Serialize};
80use serde_json::Value as JsonValue;
81#[cfg(any(feature = "ws", feature = "grpc", all(feature = "dst", reifydb_single_threaded)))]
82pub use subscription::{BatchItem, HydrationConfig, Linger, SubscriptionConfig, Throttle, build_subscription_rql};
83#[cfg(feature = "ws")]
84pub use ws::{WsBatchSubscription, WsClient, WsClientOptions};
85
86#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
87#[derive(Debug, Clone)]
88pub struct ResponseMeta {
89	pub fingerprint: String,
90	pub duration: String,
91}
92
93#[derive(Debug)]
94pub struct AdminResult {
95	pub frames: Vec<Frame>,
96	pub meta: Option<ResponseMeta>,
97}
98
99#[derive(Debug)]
100pub struct CommandResult {
101	pub frames: Vec<Frame>,
102	pub meta: Option<ResponseMeta>,
103}
104
105#[derive(Debug)]
106pub struct QueryResult {
107	pub frames: Vec<Frame>,
108	pub meta: Option<ResponseMeta>,
109}
110
111#[derive(Debug, Clone)]
112pub struct LoginResult {
113	pub token: String,
114	pub identity: String,
115}
116
117#[cfg(any(feature = "ws", feature = "grpc"))]
118pub fn connection_lost_error() -> Error {
119	ClientError::ConnectionLost.into()
120}
121
122/// Automatic-reconnection settings shared by the WebSocket and gRPC clients.
123///
124/// Backoff is `reconnect_delay_ms * 2^(attempt-1)`, capped at 30s; a successful reconnect
125/// restores the session token and re-establishes active subscriptions against the same handles.
126#[cfg(any(feature = "ws", feature = "grpc"))]
127#[derive(Clone)]
128pub struct ReconnectOptions {
129	pub max_reconnect_attempts: u32,
130	pub reconnect_delay_ms: u64,
131	pub connect_timeout_ms: u64,
132	pub on_disconnect: Option<Arc<dyn Fn() + Send + Sync>>,
133	pub on_reconnect: Option<Arc<dyn Fn() + Send + Sync>>,
134}
135
136#[cfg(any(feature = "ws", feature = "grpc"))]
137impl Default for ReconnectOptions {
138	fn default() -> Self {
139		Self {
140			max_reconnect_attempts: 5,
141			reconnect_delay_ms: 1000,
142			connect_timeout_ms: 30_000,
143			on_disconnect: None,
144			on_reconnect: None,
145		}
146	}
147}
148
149#[cfg(any(feature = "http", feature = "ws"))]
150/// Wire format for a single typed value: `{"type": "Int2", "value": "1234"}`.
151#[derive(Debug, Serialize, Deserialize)]
152pub struct WireValue {
153	#[serde(rename = "type")]
154	pub type_name: String,
155	pub value: String,
156}
157
158#[cfg(any(feature = "http", feature = "ws"))]
159/// Untagged: positional serializes as a JSON array, named as a JSON object.
160#[derive(Debug, Serialize, Deserialize)]
161#[serde(untagged)]
162pub enum WireParams {
163	Positional(Vec<WireValue>),
164	Named(HashMap<String, WireValue>),
165}
166
167#[cfg(any(feature = "http", feature = "ws"))]
168fn value_to_wire(value: Value) -> WireValue {
169	let (type_name, value_str): (&str, String) = match &value {
170		Value::None {
171			..
172		} => ("None", "\u{27EA}none\u{27EB}".to_string()),
173		Value::Boolean(b) => ("Boolean", b.to_string()),
174		Value::Float4(f) => ("Float4", f.to_string()),
175		Value::Float8(f) => ("Float8", f.to_string()),
176		Value::Int1(i) => ("Int1", i.to_string()),
177		Value::Int2(i) => ("Int2", i.to_string()),
178		Value::Int4(i) => ("Int4", i.to_string()),
179		Value::Int8(i) => ("Int8", i.to_string()),
180		Value::Int16(i) => ("Int16", i.to_string()),
181		Value::Utf8(s) => ("Utf8", s.clone()),
182		Value::Uint1(u) => ("Uint1", u.to_string()),
183		Value::Uint2(u) => ("Uint2", u.to_string()),
184		Value::Uint4(u) => ("Uint4", u.to_string()),
185		Value::Uint8(u) => ("Uint8", u.to_string()),
186		Value::Uint16(u) => ("Uint16", u.to_string()),
187		Value::Uuid4(u) => ("Uuid4", u.to_string()),
188		Value::Uuid7(u) => ("Uuid7", u.to_string()),
189		Value::Date(d) => ("Date", d.to_string()),
190		Value::DateTime(dt) => ("DateTime", dt.to_string()),
191		Value::Time(t) => ("Time", t.to_string()),
192		Value::Duration(d) => ("Duration", d.to_iso_string()),
193		Value::Blob(b) => ("Blob", b.to_hex()),
194		Value::IdentityId(id) => ("IdentityId", id.to_string()),
195		Value::Int(i) => ("Int", i.to_string()),
196		Value::Uint(u) => ("Uint", u.to_string()),
197		Value::Decimal(d) => ("Decimal", d.to_string()),
198		Value::Any(v) => return value_to_wire(*v.clone()),
199		Value::DictionaryId(id) => ("DictionaryId", id.to_string()),
200		Value::Type(t) => ("ValueType", t.to_string()),
201		Value::List(items) => ("List", format!("{}", Value::List(items.clone()))),
202		Value::Record(fields) => ("Record", format!("{}", Value::Record(fields.clone()))),
203		Value::Tuple(items) => ("Tuple", format!("{}", Value::Tuple(items.clone()))),
204	};
205	WireValue {
206		type_name: type_name.to_string(),
207		value: value_str,
208	}
209}
210
211#[cfg(any(feature = "http", feature = "ws"))]
212pub fn params_to_wire(params: Params) -> Option<WireParams> {
213	match params {
214		Params::None => None,
215		Params::Positional(values) => Some(WireParams::Positional(
216			Arc::unwrap_or_clone(values).into_iter().map(value_to_wire).collect(),
217		)),
218		Params::Named(map) => Some(WireParams::Named(
219			Arc::unwrap_or_clone(map).into_iter().map(|(k, v)| (k, value_to_wire(v))).collect(),
220		)),
221	}
222}
223
224#[cfg(any(feature = "http", feature = "ws"))]
225#[derive(Debug, Serialize, Deserialize)]
226pub struct Request {
227	pub id: String,
228	#[serde(flatten)]
229	pub payload: RequestPayload,
230}
231
232#[cfg(any(feature = "http", feature = "ws"))]
233#[derive(Debug, Serialize, Deserialize)]
234#[serde(tag = "type", content = "payload")]
235pub enum RequestPayload {
236	Auth(AuthRequest),
237	Admin(AdminRequest),
238	Command(CommandRequest),
239	Query(QueryRequest),
240	Subscribe(SubscribeRequest),
241	Unsubscribe(UnsubscribeRequest),
242	BatchSubscribe(BatchSubscribeRequest),
243	BatchUnsubscribe(BatchUnsubscribeRequest),
244	Call(CallRequest),
245	Logout,
246}
247
248#[cfg(any(feature = "http", feature = "ws"))]
249#[derive(Debug, Serialize, Deserialize)]
250pub struct AdminRequest {
251	pub rql: String,
252	pub params: Option<WireParams>,
253	#[serde(skip_serializing_if = "Option::is_none")]
254	pub format: Option<WireFormat>,
255}
256
257#[cfg(any(feature = "http", feature = "ws"))]
258#[derive(Debug, Serialize, Deserialize)]
259pub struct AuthRequest {
260	#[serde(skip_serializing_if = "Option::is_none")]
261	pub token: Option<String>,
262	#[serde(skip_serializing_if = "Option::is_none")]
263	pub method: Option<String>,
264	#[serde(skip_serializing_if = "Option::is_none")]
265	pub credentials: Option<HashMap<String, String>>,
266}
267
268#[cfg(any(feature = "http", feature = "ws"))]
269#[derive(Debug, Serialize, Deserialize)]
270pub struct CommandRequest {
271	pub rql: String,
272	pub params: Option<WireParams>,
273	#[serde(skip_serializing_if = "Option::is_none")]
274	pub format: Option<WireFormat>,
275}
276
277#[cfg(any(feature = "http", feature = "ws"))]
278#[derive(Debug, Serialize, Deserialize)]
279pub struct QueryRequest {
280	pub rql: String,
281	pub params: Option<WireParams>,
282	#[serde(skip_serializing_if = "Option::is_none")]
283	pub format: Option<WireFormat>,
284}
285
286#[cfg(any(feature = "http", feature = "ws"))]
287#[derive(Debug, Serialize, Deserialize)]
288pub struct SubscribeRequest {
289	pub rql: String,
290	#[serde(skip_serializing_if = "Option::is_none")]
291	pub format: Option<WireFormat>,
292}
293
294#[cfg(any(feature = "http", feature = "ws"))]
295#[derive(Debug, Serialize, Deserialize)]
296pub struct UnsubscribeRequest {
297	pub subscription_id: String,
298}
299
300#[cfg(any(feature = "http", feature = "ws"))]
301#[derive(Debug, Serialize, Deserialize)]
302pub struct BatchSubscribeRequest {
303	pub queries: Vec<String>,
304	#[serde(skip_serializing_if = "Option::is_none")]
305	pub format: Option<WireFormat>,
306}
307
308#[cfg(any(feature = "http", feature = "ws"))]
309#[derive(Debug, Serialize, Deserialize)]
310pub struct BatchUnsubscribeRequest {
311	pub batch_id: String,
312}
313
314#[cfg(any(feature = "http", feature = "ws"))]
315#[derive(Debug, Serialize, Deserialize)]
316pub struct CallRequest {
317	pub name: String,
318	pub params: Option<WireParams>,
319	#[serde(skip_serializing_if = "Option::is_none")]
320	pub format: Option<WireFormat>,
321}
322
323#[cfg(any(feature = "http", feature = "ws"))]
324#[derive(Debug, Serialize, Deserialize)]
325pub struct Response {
326	pub id: String,
327	#[serde(flatten)]
328	pub payload: ResponsePayload,
329}
330
331#[cfg(any(feature = "http", feature = "ws"))]
332#[derive(Debug, Serialize, Deserialize)]
333#[serde(tag = "type", content = "payload")]
334pub enum ResponsePayload {
335	Auth(AuthResponse),
336	Err(ErrResponse),
337	Admin(AdminResponse),
338	Command(CommandResponse),
339	Query(QueryResponse),
340	Subscribed(SubscribedResponse),
341	Unsubscribed(UnsubscribedResponse),
342	BatchSubscribed(BatchSubscribedResponse),
343	BatchUnsubscribed(BatchUnsubscribedResponse),
344	Call(CallResponse),
345	Logout(LogoutResponsePayload),
346}
347
348#[cfg(any(feature = "http", feature = "ws"))]
349#[derive(Debug, Serialize, Deserialize)]
350pub struct AdminResponse {
351	pub content_type: String,
352	pub body: JsonValue,
353	#[serde(default)]
354	pub meta: Option<ResponseMeta>,
355}
356
357#[cfg(any(feature = "http", feature = "ws"))]
358use reifydb_value::error::Diagnostic;
359
360#[cfg(any(feature = "http", feature = "ws"))]
361#[derive(Debug, Serialize, Deserialize)]
362pub struct AuthResponse {
363	#[serde(skip_serializing_if = "Option::is_none")]
364	pub status: Option<String>,
365	#[serde(skip_serializing_if = "Option::is_none")]
366	pub token: Option<String>,
367	#[serde(skip_serializing_if = "Option::is_none")]
368	pub identity: Option<String>,
369}
370
371#[cfg(any(feature = "http", feature = "ws"))]
372#[derive(Debug, Serialize, Deserialize)]
373pub struct ErrResponse {
374	pub diagnostic: Diagnostic,
375}
376
377#[cfg(any(feature = "http", feature = "ws"))]
378#[derive(Debug, Serialize, Deserialize)]
379pub struct CommandResponse {
380	pub content_type: String,
381	pub body: JsonValue,
382	#[serde(default)]
383	pub meta: Option<ResponseMeta>,
384}
385
386#[cfg(any(feature = "http", feature = "ws"))]
387#[derive(Debug, Serialize, Deserialize)]
388pub struct QueryResponse {
389	pub content_type: String,
390	pub body: JsonValue,
391	#[serde(default)]
392	pub meta: Option<ResponseMeta>,
393}
394
395#[cfg(any(feature = "http", feature = "ws"))]
396#[derive(Debug, Serialize, Deserialize)]
397pub struct CallResponse {
398	pub content_type: String,
399	pub body: JsonValue,
400	#[serde(default)]
401	pub meta: Option<ResponseMeta>,
402}
403
404#[cfg(any(feature = "http", feature = "ws"))]
405#[derive(Debug, Serialize, Deserialize)]
406pub struct SubscribedResponse {
407	pub subscription_id: String,
408}
409
410#[cfg(any(feature = "http", feature = "ws"))]
411#[derive(Debug, Serialize, Deserialize)]
412pub struct UnsubscribedResponse {
413	pub subscription_id: String,
414}
415
416#[cfg(any(feature = "http", feature = "ws"))]
417#[derive(Debug, Serialize, Deserialize)]
418pub struct BatchSubscribedResponse {
419	pub batch_id: String,
420	pub members: Vec<BatchMemberInfo>,
421}
422
423#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
424#[derive(Debug, Clone)]
425pub struct BatchMemberInfo {
426	pub index: usize,
427	pub subscription_id: String,
428}
429
430#[cfg(any(feature = "http", feature = "ws"))]
431#[derive(Debug, Serialize, Deserialize)]
432pub struct BatchUnsubscribedResponse {
433	pub batch_id: String,
434}
435
436#[cfg(any(feature = "http", feature = "ws"))]
437#[derive(Debug, Serialize, Deserialize)]
438pub struct LogoutResponsePayload {
439	pub status: String,
440}
441
442#[cfg(any(feature = "http", feature = "ws"))]
443#[derive(Debug, Serialize, Deserialize)]
444#[serde(tag = "type", content = "payload")]
445pub enum ServerPush {
446	Change(WireChangePayload),
447	BatchChange(WireBatchChangePayload),
448	BatchMemberClosed(BatchMemberClosedPayload),
449	BatchClosed(BatchClosedPayload),
450}
451
452#[cfg(any(feature = "http", feature = "ws"))]
453#[derive(Debug, Clone, Serialize, Deserialize)]
454pub struct WireChangePayload {
455	pub subscription_id: String,
456	pub content_type: String,
457	pub body: JsonValue,
458}
459
460#[cfg(any(feature = "http", feature = "ws"))]
461#[derive(Debug, Clone, Serialize, Deserialize)]
462pub struct WireBatchChangePayload {
463	pub batch_id: String,
464	pub entries: Vec<WireBatchChangeEntry>,
465}
466
467#[cfg(any(feature = "http", feature = "ws"))]
468#[derive(Debug, Clone, Serialize, Deserialize)]
469pub struct WireBatchChangeEntry {
470	pub subscription_id: String,
471	pub content_type: String,
472	pub body: JsonValue,
473}
474
475#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
476#[derive(Debug, Clone, Copy, PartialEq, Eq)]
477pub enum ChangeKind {
478	Insert,
479	Update,
480	Remove,
481}
482
483/// The server stages one uniform `_op` per frame, so a change carries exactly one kind;
484/// the `_op` column is already stripped from `frame`.
485#[derive(Debug, Clone)]
486pub struct FrameChange {
487	pub kind: ChangeKind,
488	pub frame: Frame,
489}
490
491#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
492#[derive(Debug, Clone)]
493pub struct ChangePayload {
494	pub subscription_id: String,
495	pub content_type: String,
496	pub body: JsonValue,
497	#[cfg_attr(any(feature = "http", feature = "ws"), serde(skip, default))]
498	pub changes: Vec<FrameChange>,
499}
500
501#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
502#[derive(Debug, Clone)]
503pub struct BatchChangePayload {
504	pub batch_id: String,
505	pub entries: Vec<BatchChangeEntry>,
506}
507
508#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
509#[derive(Debug, Clone)]
510pub struct BatchChangeEntry {
511	pub subscription_id: String,
512	pub content_type: String,
513	pub body: JsonValue,
514	#[cfg_attr(any(feature = "http", feature = "ws"), serde(skip, default))]
515	pub changes: Vec<FrameChange>,
516	#[cfg_attr(any(feature = "http", feature = "ws"), serde(skip, default))]
517	pub decode_error: Option<String>,
518}
519
520#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
521#[derive(Debug, Clone)]
522pub struct BatchMemberClosedPayload {
523	pub batch_id: String,
524	pub subscription_id: String,
525}
526
527#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
528#[derive(Debug, Clone)]
529pub struct BatchClosedPayload {
530	pub batch_id: String,
531}
532
533#[derive(Debug, Clone)]
534pub enum BatchPushEvent {
535	Change(BatchChangePayload),
536	MemberClosed(BatchMemberClosedPayload),
537	Closed(BatchClosedPayload),
538}
reifydb_client/lib.rs

reifydb_client/
lib.rs
