reifydb_sub_server/
response.rs1use reifydb_codec::{
5 frame::{encode::encode_frames, options::EncodeOptions},
6 json::{none_marker, wire_type::to_json as type_to_json},
7};
8use reifydb_core::value::batch::views;
9use reifydb_value::{
10 reifydb_assertions,
11 value::{
12 Value,
13 column_view::ColumnView,
14 diff_type::DiffType,
15 frame::frame::Frame,
16 system_columns::{SystemColumn, row_numbers},
17 value_type::ValueType,
18 },
19};
20use serde_json::{self, Map, Value as JsonValue, to_string as json_to_string};
21
22pub const CONTENT_TYPE_JSON: &str = "application/vnd.reifydb.json";
23pub const CONTENT_TYPE_FRAMES: &str = "application/vnd.reifydb.frames";
24pub const CONTENT_TYPE_RBCF: &str = "application/vnd.reifydb.rbcf";
25
26pub fn encode_frames_rbcf(frames: &[Frame]) -> Result<Vec<u8>, String> {
27 encode_frames(frames, &EncodeOptions::fast()).map_err(|e| e.to_string())
28}
29
30pub struct ResolvedResponse {
31 pub content_type: String,
32 pub body: String,
33}
34
35pub fn resolve_change_json(frames: Vec<Frame>) -> Result<ResolvedResponse, String> {
36 let envelopes: Vec<JsonValue> = frames.iter().map(change_envelope).collect::<Result<_, String>>()?;
37 Ok(json_response(json_to_string(&envelopes).map_err(|e| e.to_string())?))
38}
39
40fn change_envelope(frame: &Frame) -> Result<JsonValue, String> {
41 let rows = frame_json_rows(frame)?;
42
43 let mut envelope = Map::new();
44 if let Some(op) = frame.op {
45 envelope.insert("op".to_string(), JsonValue::from(DiffType::as_u8(op)));
46 }
47 envelope.insert("types".to_string(), frame_types(frame)?);
48 envelope.insert("rows".to_string(), JsonValue::Array(rows));
49 Ok(JsonValue::Object(envelope))
50}
51
52pub fn resolve_response_json(frames: Vec<Frame>, unwrap: bool) -> Result<ResolvedResponse, String> {
53 if frames.is_empty() {
54 return Ok(json_response("[]".to_string()));
55 }
56
57 if has_body_column(&frames) {
58 let frame = frames.into_iter().next().unwrap();
59 return Ok(json_response(render_body_column(frame, unwrap)?));
60 }
61
62 Ok(json_response(render_frame_rows(&frames, unwrap)?))
63}
64
65#[inline]
66fn json_response(body: String) -> ResolvedResponse {
67 ResolvedResponse {
68 content_type: CONTENT_TYPE_JSON.to_string(),
69 body,
70 }
71}
72
73#[inline]
74fn has_body_column(frames: &[Frame]) -> bool {
75 frames.first().map(|f| f.batch.schema_ref().column_with_name("body").is_some()).unwrap_or(false)
76}
77
78#[inline]
79fn render_body_column(frame: Frame, unwrap: bool) -> Result<String, String> {
80 let body_col = frame.try_column("body").map_err(|e| e.to_string())?;
81
82 let row_count = body_col.len();
83 if body_col.is_utf8() {
84 let values: Vec<String> = (0..row_count).map(|i| body_col.as_string(i)).collect();
85 if unwrap || values.len() == 1 {
86 Ok(values.into_iter().next().unwrap())
87 } else {
88 Ok(format!("[{}]", values.join(", ")))
89 }
90 } else {
91 let json_values: Vec<JsonValue> =
92 (0..row_count).map(|i| body_col.get_value(i).to_json_value()).collect();
93 reifydb_assertions! {
94 let len = json_values.len();
95 assert!(
96 len > 0,
97 "render_body_column reached the non-utf8 branch with an empty body column, but the \
98 caller already established the frame is non-empty by detecting a body column on it; \
99 indexing json_values[0] would panic on a row_count={len} column"
100 );
101 }
102 if unwrap {
103 Ok(json_to_string(&json_values[0]).unwrap())
104 } else {
105 Ok(json_to_string(&json_values).unwrap())
106 }
107 }
108}
109
110fn option_depth(ty: &ValueType) -> u32 {
111 let mut depth = 0;
112 let mut base = ty;
113 while let ValueType::Option(inner) = base {
114 depth += 1;
115 base = inner;
116 }
117 depth
118}
119
120fn row_value(column_type: &ValueType, value: Value) -> JsonValue {
121 match &value {
122 Value::None {
123 inner,
124 } => {
125 let wrapped = option_depth(column_type).saturating_sub(option_depth(inner) + 1);
126 if wrapped == 0 {
127 JsonValue::Null
128 } else {
129 JsonValue::String(none_marker(wrapped))
130 }
131 }
132 _ => value.to_json_value(),
133 }
134}
135
136fn frame_types(frame: &Frame) -> Result<JsonValue, String> {
137 let mut types = Map::new();
138 for view in column_views(frame)? {
139 types.insert(view.field.name().clone(), type_to_json(&view.get_type()));
140 }
141 Ok(JsonValue::Object(types))
142}
143
144fn column_views(frame: &Frame) -> Result<Vec<ColumnView<'_>>, String> {
145 let mut columns = views(&frame.batch).map_err(|e| e.to_string())?;
146 columns.retain(|view| view.field.name() != SystemColumn::RowNumbers.name());
147 Ok(columns)
148}
149
150#[inline]
151fn render_frame_rows(frames: &[Frame], unwrap: bool) -> Result<String, String> {
152 let json_frames = frames_to_json_rows(frames)?;
153
154 if unwrap && json_frames.len() == 1 && json_frames[0].len() == 1 {
155 return Ok(json_to_string(&json_frames[0][0]).unwrap());
156 }
157
158 let envelopes: Vec<JsonValue> = frames
159 .iter()
160 .zip(json_frames)
161 .map(|(frame, rows)| {
162 let mut envelope = Map::new();
163 envelope.insert("types".to_string(), frame_types(frame)?);
164 envelope.insert("rows".to_string(), JsonValue::Array(rows));
165 Ok(JsonValue::Object(envelope))
166 })
167 .collect::<Result<_, String>>()?;
168 Ok(json_to_string(&envelopes).unwrap())
169}
170
171fn frames_to_json_rows(frames: &[Frame]) -> Result<Vec<Vec<JsonValue>>, String> {
172 frames.iter().map(frame_json_rows).collect()
173}
174
175fn frame_json_rows(frame: &Frame) -> Result<Vec<JsonValue>, String> {
176 let numbers = row_numbers(&frame.batch).map_err(|e| e.to_string())?;
177 let columns = column_views(frame)?;
178 Ok((0..frame.batch.num_rows())
179 .map(|i| {
180 let mut obj = Map::new();
181 if let Some(rn) = numbers.get(i) {
182 obj.insert(SystemColumn::RowNumbers.name().to_string(), JsonValue::from(rn.value()));
183 }
184 for view in &columns {
185 obj.insert(view.field.name().clone(), row_value(&view.get_type(), view.get_value(i)));
186 }
187 JsonValue::Object(obj)
188 })
189 .collect())
190}