use serde_json::{Map, Value};
use crate::model::{Frame, Message, MsgKind, OpResult};
use crate::segment::{ColumnValue, RowsSegment};
use crate::util::base64;
fn obj(entries: Vec<(&str, Value)>) -> Value {
let mut map = Map::new();
for (k, v) in entries {
map.insert(k.to_string(), v);
}
Value::Object(map)
}
fn scope_map_value(scopes: &[(String, Vec<String>)]) -> Value {
let mut map = Map::new();
for (k, vs) in scopes {
map.insert(
k.clone(),
Value::Array(vs.iter().map(|v| Value::String(v.clone())).collect()),
);
}
Value::Object(map)
}
pub fn render_message(msg: &Message) -> Value {
obj(vec![
("magic", Value::from("SSP2")),
("wireVersion", Value::from(1)),
(
"msgKind",
Value::from(match msg.msg_kind {
MsgKind::Request => "request",
MsgKind::Response => "response",
}),
),
(
"frames",
Value::Array(msg.frames.iter().map(render_frame).collect()),
),
])
}
fn render_frame(frame: &Frame) -> Value {
let mut entries: Vec<(&str, Value)> = Vec::new();
match frame {
Frame::ReqHeader {
client_id,
schema_version,
} => {
entries.push(("type", Value::from("REQ_HEADER")));
entries.push(("clientId", Value::from(client_id.clone())));
entries.push(("schemaVersion", Value::from(*schema_version)));
}
Frame::PushCommit {
client_commit_id,
operations,
} => {
entries.push(("type", Value::from("PUSH_COMMIT")));
entries.push(("clientCommitId", Value::from(client_commit_id.clone())));
let ops = operations
.iter()
.map(|op| {
let mut e: Vec<(&str, Value)> = vec![
("table", Value::from(op.table.clone())),
("rowId", Value::from(op.row_id.clone())),
("op", Value::from(op.op.name())),
];
if let Some(v) = op.base_version {
e.push(("baseVersion", Value::from(v)));
}
if let Some(p) = &op.payload {
e.push(("payload", Value::from(base64(p))));
}
obj(e)
})
.collect();
entries.push(("operations", Value::Array(ops)));
}
Frame::PullHeader {
limit_commits,
limit_snapshot_rows,
max_snapshot_pages,
accept,
} => {
entries.push(("type", Value::from("PULL_HEADER")));
entries.push(("limitCommits", Value::from(*limit_commits)));
entries.push(("limitSnapshotRows", Value::from(*limit_snapshot_rows)));
entries.push(("maxSnapshotPages", Value::from(*max_snapshot_pages)));
entries.push(("accept", Value::from(*accept)));
}
Frame::Subscription {
id,
table,
scopes,
params,
cursor,
bootstrap_state,
} => {
entries.push(("type", Value::from("SUBSCRIPTION")));
entries.push(("id", Value::from(id.clone())));
entries.push(("table", Value::from(table.clone())));
entries.push(("scopes", scope_map_value(scopes)));
if let Some(p) = params {
entries.push(("params", p.parse()));
}
entries.push(("cursor", Value::from(*cursor)));
if let Some(b) = bootstrap_state {
entries.push(("bootstrapState", b.parse()));
}
}
Frame::RespHeader {
required_schema_version,
latest_schema_version,
} => {
entries.push(("type", Value::from("RESP_HEADER")));
if let Some(v) = required_schema_version {
entries.push(("requiredSchemaVersion", Value::from(*v)));
}
if let Some(v) = latest_schema_version {
entries.push(("latestSchemaVersion", Value::from(*v)));
}
}
Frame::Lease {
lease_id,
expires_at_ms,
} => {
entries.push(("type", Value::from("LEASE")));
entries.push(("leaseId", Value::from(lease_id.clone())));
entries.push(("expiresAtMs", Value::from(*expires_at_ms)));
}
Frame::PushResult {
client_commit_id,
status,
commit_seq,
results,
} => {
entries.push(("type", Value::from("PUSH_RESULT")));
entries.push(("clientCommitId", Value::from(client_commit_id.clone())));
entries.push(("status", Value::from(status.name())));
if let Some(v) = commit_seq {
entries.push(("commitSeq", Value::from(*v)));
}
entries.push((
"results",
Value::Array(results.iter().map(render_result).collect()),
));
}
Frame::SubStart {
id,
status,
reason_code,
effective_scopes,
bootstrap,
} => {
entries.push(("type", Value::from("SUB_START")));
entries.push(("id", Value::from(id.clone())));
entries.push(("status", Value::from(status.name())));
entries.push(("reasonCode", Value::from(reason_code.clone())));
entries.push(("effectiveScopes", scope_map_value(effective_scopes)));
entries.push(("bootstrap", Value::from(*bootstrap)));
}
Frame::Commit {
commit_seq,
created_at_ms,
actor_id,
tables,
changes,
} => {
entries.push(("type", Value::from("COMMIT")));
entries.push(("commitSeq", Value::from(*commit_seq)));
entries.push(("createdAtMs", Value::from(*created_at_ms)));
entries.push(("actorId", Value::from(actor_id.clone())));
entries.push((
"tables",
Value::Array(tables.iter().map(|t| Value::from(t.clone())).collect()),
));
let rendered_changes = changes
.iter()
.map(|c| {
let mut e: Vec<(&str, Value)> = vec![
("tableIndex", Value::from(c.table_index)),
("rowId", Value::from(c.row_id.clone())),
("op", Value::from(c.op.name())),
];
if let Some(v) = c.row_version {
e.push(("rowVersion", Value::from(v)));
}
let mut scopes = Map::new();
for (k, v) in &c.scopes {
scopes.insert(k.clone(), Value::from(v.clone()));
}
e.push(("scopes", Value::Object(scopes)));
if let Some(row) = &c.row {
e.push(("row", Value::from(base64(row))));
}
obj(e)
})
.collect();
entries.push(("changes", Value::Array(rendered_changes)));
}
Frame::SegmentRef {
segment_id,
media_type,
table,
byte_length,
row_count,
as_of_commit_seq,
scope_digest,
row_cursor,
next_row_cursor,
url,
url_expires_at_ms,
} => {
entries.push(("type", Value::from("SEGMENT_REF")));
entries.push(("segmentId", Value::from(segment_id.clone())));
entries.push(("mediaType", Value::from(media_type.name())));
entries.push(("table", Value::from(table.clone())));
entries.push(("byteLength", Value::from(*byte_length)));
entries.push(("rowCount", Value::from(*row_count)));
entries.push(("asOfCommitSeq", Value::from(*as_of_commit_seq)));
entries.push(("scopeDigest", Value::from(scope_digest.clone())));
if let Some(v) = row_cursor {
entries.push(("rowCursor", Value::from(v.clone())));
}
if let Some(v) = next_row_cursor {
entries.push(("nextRowCursor", Value::from(v.clone())));
}
if let Some(v) = url {
entries.push(("url", Value::from(v.clone())));
}
if let Some(v) = url_expires_at_ms {
entries.push(("urlExpiresAtMs", Value::from(*v)));
}
}
Frame::SegmentInline { payload } => {
entries.push(("type", Value::from("SEGMENT_INLINE")));
entries.push(("payload", Value::from(base64(payload))));
}
Frame::SubEnd {
next_cursor,
bootstrap_state,
} => {
entries.push(("type", Value::from("SUB_END")));
entries.push(("nextCursor", Value::from(*next_cursor)));
if let Some(b) = bootstrap_state {
entries.push(("bootstrapState", b.parse()));
}
}
Frame::Error {
code,
message,
category,
retryable,
recommended_action,
details,
} => {
entries.push(("type", Value::from("ERROR")));
entries.push(("code", Value::from(code.clone())));
entries.push(("message", Value::from(message.clone())));
entries.push(("category", Value::from(category.clone())));
entries.push(("retryable", Value::from(*retryable)));
entries.push(("recommendedAction", Value::from(recommended_action.clone())));
if let Some(d) = details {
entries.push(("details", d.parse()));
}
}
Frame::Unknown {
frame_type,
payload,
} => {
entries.push(("type", Value::from("UNKNOWN")));
entries.push(("frameType", Value::from(*frame_type)));
entries.push(("payload", Value::from(base64(payload))));
}
}
obj(entries)
}
fn render_result(result: &OpResult) -> Value {
match result {
OpResult::Applied { op_index } => obj(vec![
("opIndex", Value::from(*op_index)),
("status", Value::from("applied")),
]),
OpResult::Conflict {
op_index,
code,
message,
server_version,
server_row,
} => obj(vec![
("opIndex", Value::from(*op_index)),
("status", Value::from("conflict")),
("code", Value::from(code.clone())),
("message", Value::from(message.clone())),
("serverVersion", Value::from(*server_version)),
("serverRow", Value::from(base64(server_row))),
]),
OpResult::Error {
op_index,
code,
message,
retryable,
} => obj(vec![
("opIndex", Value::from(*op_index)),
("status", Value::from("error")),
("code", Value::from(code.clone())),
("message", Value::from(message.clone())),
("retryable", Value::from(*retryable)),
]),
}
}
pub fn render_rows_segment(seg: &RowsSegment) -> Value {
let columns = seg
.columns
.iter()
.map(|c| {
obj(vec![
("name", Value::from(c.name.clone())),
("type", Value::from(c.ty.name())),
("nullable", Value::from(c.nullable)),
])
})
.collect();
let blocks = seg
.blocks
.iter()
.map(|block| {
Value::Array(
block
.iter()
.map(|row| {
let mut values = Map::new();
for (col, value) in seg.columns.iter().zip(row.values.iter()) {
values.insert(col.name.clone(), render_column_value(value));
}
obj(vec![
("serverVersion", Value::from(row.server_version)),
("values", Value::Object(values)),
])
})
.collect(),
)
})
.collect();
obj(vec![
("magic", Value::from("SSG2")),
("formatVersion", Value::from(1)),
("table", Value::from(seg.table.clone())),
("schemaVersion", Value::from(seg.schema_version)),
("columns", Value::Array(columns)),
("blocks", Value::Array(blocks)),
])
}
fn render_column_value(value: &Option<ColumnValue>) -> Value {
match value {
None => Value::Null,
Some(ColumnValue::String(s)) => Value::from(s.clone()),
Some(ColumnValue::Integer(v)) => Value::from(*v),
Some(ColumnValue::Float(v)) => {
serde_json::Number::from_f64(*v).map_or(Value::Null, Value::Number)
}
Some(ColumnValue::Boolean(v)) => Value::from(*v),
Some(ColumnValue::Json(j)) => j.parse(),
Some(ColumnValue::Bytes(b)) => Value::from(base64(b)),
Some(ColumnValue::BlobRef(j)) => j.parse(),
Some(ColumnValue::Crdt(b)) => Value::from(base64(b)),
}
}