use std::collections::BTreeMap;
use hyphae_query::{
AggregationPlan, AggregationResult, CompareOperator, Cursor, FieldPath, Filter, GroupResult,
Metric, MetricValue, NamedMetric, NamedMetricValue, NullPlacement, Query, QueryResult, Record,
SortDirection, SortField, Value,
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use thiserror::Error;
pub const BYTES_HEX_KEY: &str = "$hyphae_bytes_hex";
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ErrorV1 {
pub code: String,
pub message: String,
#[schemars(with = "uuid::Uuid")]
pub request_id: String,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CapabilitiesV1 {
pub api_version: String,
pub disk_format_version: u16,
pub features: Vec<String>,
pub limits: ApiLimitsV1,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ApiLimitsV1 {
pub key_bytes: u64,
pub document_bytes: u64,
pub request_body_bytes: u64,
pub json_depth: u64,
pub json_nodes: u64,
pub request_body_timeout_ms: u64,
pub batch_items: u64,
pub scanned_records: u64,
pub matched_records: u64,
pub result_rows: u64,
pub aggregation_groups: u64,
pub filter_nodes: u64,
pub filter_depth: u64,
pub sort_fields: u64,
pub group_fields: u64,
pub metrics: u64,
pub concurrent_operations: u64,
pub query_timeout_ms: u64,
pub proof_bytes: u64,
pub witness_bytes: u64,
pub response_bytes: u64,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct HealthV1 {
pub status: String,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RecordV1 {
pub key_hex: String,
pub value: serde_json::Value,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PutRequestV1 {
#[schemars(with = "Option<uuid::Uuid>")]
pub transaction_id: Option<String>,
pub records: Vec<RecordV1>,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct DeleteRequestV1 {
#[schemars(with = "Option<uuid::Uuid>")]
pub transaction_id: Option<String>,
pub keys_hex: Vec<String>,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct GetRequestV1 {
pub key_hex: String,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CommitReceiptV1 {
pub status: String,
#[schemars(with = "uuid::Uuid")]
pub transaction_id: String,
pub commit_sequence: u64,
pub commit_digest: String,
pub transaction_digest: String,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProofV1 {
pub encoding: String,
pub data: String,
pub proof_digest: String,
pub anchor_digest: String,
pub checkpoint_sequence: u64,
pub checkpoint_digest: Option<String>,
pub snapshot_digest: String,
pub witness: WitnessV1,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WitnessV1 {
pub path: String,
pub file_bytes: u64,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct GetResponseV1 {
pub found: bool,
pub record: Option<RecordV1>,
pub proof: ProofV1,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CompareOperatorV1 {
Equal,
NotEqual,
Less,
LessOrEqual,
Greater,
GreaterOrEqual,
}
#[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)]
pub enum FilterV1 {
#[default]
MatchAll,
Exists {
path: Vec<String>,
},
Compare {
path: Vec<String>,
operator: CompareOperatorV1,
value: serde_json::Value,
},
Prefix {
path: Vec<String>,
prefix: serde_json::Value,
},
Contains {
path: Vec<String>,
needle: serde_json::Value,
},
All {
filters: Vec<Self>,
},
Any {
filters: Vec<Self>,
},
Not {
filter: Box<Self>,
},
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SortDirectionV1 {
Ascending,
Descending,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum NullPlacementV1 {
First,
Last,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SortFieldV1 {
pub path: Vec<String>,
pub direction: SortDirectionV1,
pub nulls: NullPlacementV1,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CursorV1 {
pub sort_values: Vec<Option<serde_json::Value>>,
pub key_hex: String,
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(tag = "metric", rename_all = "snake_case", deny_unknown_fields)]
pub enum NamedMetricV1 {
Count {
name: String,
},
Sum {
name: String,
path: Vec<String>,
},
Min {
name: String,
path: Vec<String>,
},
Max {
name: String,
path: Vec<String>,
},
}
#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AggregationPlanV1 {
pub group_by: Vec<Vec<String>>,
pub metrics: Vec<NamedMetricV1>,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct QueryRequestV1 {
#[serde(default)]
pub filter: FilterV1,
#[serde(default)]
pub sort: Vec<SortFieldV1>,
pub cursor: Option<CursorV1>,
pub limit: u32,
pub aggregation: Option<AggregationPlanV1>,
pub timeout_ms: Option<u64>,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum MetricValueV1 {
Count {
value: u64,
},
Integer {
value: Option<String>,
},
Value {
value: Option<serde_json::Value>,
},
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NamedMetricResultV1 {
pub name: String,
pub result: MetricValueV1,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum GroupKeyValueV1 {
Missing,
Value {
value: serde_json::Value,
},
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct GroupResultV1 {
pub key: Vec<GroupKeyValueV1>,
pub metrics: Vec<NamedMetricResultV1>,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct AggregationResultV1 {
pub grouped: bool,
pub groups: Vec<GroupResultV1>,
}
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct QueryResponseV1 {
pub rows: Vec<RecordV1>,
pub next_cursor: Option<CursorV1>,
pub aggregation: Option<AggregationResultV1>,
pub scanned_records: u64,
pub matched_records: u64,
pub proof: ProofV1,
}
#[derive(Clone, Debug, Error, Eq, PartialEq)]
pub enum WireValueError {
#[error("structured values accept only signed 64-bit integer JSON numbers")]
NonIntegerNumber,
#[error("invalid reserved bytes hexadecimal envelope")]
InvalidBytesHex,
#[error("invalid binary key hexadecimal")]
InvalidKeyHex,
#[error("field path contains an empty segment")]
EmptyPathSegment,
#[error("query page size is not representable on this target")]
LengthOverflow,
}
impl QueryRequestV1 {
pub fn to_domain(&self) -> Result<Query, WireValueError> {
Ok(Query {
filter: filter_to_domain(&self.filter)?,
sort: self
.sort
.iter()
.map(sort_to_domain)
.collect::<Result<_, _>>()?,
cursor: self.cursor.as_ref().map(cursor_to_domain).transpose()?,
limit: usize::try_from(self.limit).map_err(|_| WireValueError::LengthOverflow)?,
aggregation: self
.aggregation
.as_ref()
.map(aggregation_to_domain)
.transpose()?,
})
}
}
impl RecordV1 {
pub fn to_domain(&self) -> Result<Record, WireValueError> {
Ok(Record {
key: decode_key_hex(&self.key_hex)?,
value: value_to_domain(&self.value)?,
})
}
pub fn from_domain(record: &Record) -> Self {
Self {
key_hex: encode_hex(&record.key),
value: value_from_domain(&record.value),
}
}
}
impl QueryResponseV1 {
pub fn from_domain(result: &QueryResult, proof: ProofV1) -> Self {
Self {
rows: result.rows.iter().map(RecordV1::from_domain).collect(),
next_cursor: result.next_cursor.as_ref().map(cursor_from_domain),
aggregation: result.aggregation.as_ref().map(aggregation_from_domain),
scanned_records: result.scanned_records,
matched_records: result.matched_records,
proof,
}
}
}
pub fn value_to_domain(value: &serde_json::Value) -> Result<Value, WireValueError> {
match value {
serde_json::Value::Null => Ok(Value::Null),
serde_json::Value::Bool(value) => Ok(Value::Boolean(*value)),
serde_json::Value::Number(value) => value
.as_i64()
.map(Value::Integer)
.ok_or(WireValueError::NonIntegerNumber),
serde_json::Value::String(value) => Ok(Value::String(value.clone())),
serde_json::Value::Array(values) => values
.iter()
.map(value_to_domain)
.collect::<Result<Vec<_>, _>>()
.map(Value::Array),
serde_json::Value::Object(values) => {
if values.len() == 1
&& let Some(serde_json::Value::String(encoded)) = values.get(BYTES_HEX_KEY)
{
return decode_hex(encoded)
.map(Value::Bytes)
.ok_or(WireValueError::InvalidBytesHex);
}
values
.iter()
.map(|(key, value)| Ok((key.clone(), value_to_domain(value)?)))
.collect::<Result<BTreeMap<_, _>, _>>()
.map(Value::Object)
}
}
}
pub fn value_from_domain(value: &Value) -> serde_json::Value {
match value {
Value::Null => serde_json::Value::Null,
Value::Boolean(value) => serde_json::Value::Bool(*value),
Value::Integer(value) => serde_json::Value::Number((*value).into()),
Value::String(value) => serde_json::Value::String(value.clone()),
Value::Bytes(value) => serde_json::json!({ BYTES_HEX_KEY: encode_hex(value) }),
Value::Array(values) => {
serde_json::Value::Array(values.iter().map(value_from_domain).collect())
}
Value::Object(values) => serde_json::Value::Object(
values
.iter()
.map(|(key, value)| (key.clone(), value_from_domain(value)))
.collect(),
),
}
}
pub fn decode_key_hex(encoded: &str) -> Result<Vec<u8>, WireValueError> {
let decoded = decode_hex(encoded).ok_or(WireValueError::InvalidKeyHex)?;
if decoded.is_empty() {
return Err(WireValueError::InvalidKeyHex);
}
Ok(decoded)
}
pub fn encode_hex(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut encoded = String::with_capacity(bytes.len().saturating_mul(2));
for byte in bytes {
encoded.push(char::from(HEX[usize::from(byte >> 4)]));
encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
}
encoded
}
fn decode_hex(encoded: &str) -> Option<Vec<u8>> {
if !encoded.len().is_multiple_of(2) {
return None;
}
encoded
.as_bytes()
.chunks_exact(2)
.map(|pair| Some((hex_nibble(pair[0])? << 4) | hex_nibble(pair[1])?))
.collect()
}
fn hex_nibble(value: u8) -> Option<u8> {
match value {
b'0'..=b'9' => Some(value - b'0'),
b'a'..=b'f' => Some(value - b'a' + 10),
b'A'..=b'F' => Some(value - b'A' + 10),
_ => None,
}
}
fn field_path(segments: &[String]) -> Result<FieldPath, WireValueError> {
if segments.iter().any(String::is_empty) {
return Err(WireValueError::EmptyPathSegment);
}
Ok(FieldPath::new(segments.iter().cloned()))
}
fn filter_to_domain(filter: &FilterV1) -> Result<Filter, WireValueError> {
match filter {
FilterV1::MatchAll => Ok(Filter::MatchAll),
FilterV1::Exists { path } => Ok(Filter::Exists(field_path(path)?)),
FilterV1::Compare {
path,
operator,
value,
} => Ok(Filter::Compare {
path: field_path(path)?,
operator: match operator {
CompareOperatorV1::Equal => CompareOperator::Equal,
CompareOperatorV1::NotEqual => CompareOperator::NotEqual,
CompareOperatorV1::Less => CompareOperator::Less,
CompareOperatorV1::LessOrEqual => CompareOperator::LessOrEqual,
CompareOperatorV1::Greater => CompareOperator::Greater,
CompareOperatorV1::GreaterOrEqual => CompareOperator::GreaterOrEqual,
},
value: value_to_domain(value)?,
}),
FilterV1::Prefix { path, prefix } => Ok(Filter::Prefix {
path: field_path(path)?,
prefix: value_to_domain(prefix)?,
}),
FilterV1::Contains { path, needle } => Ok(Filter::Contains {
path: field_path(path)?,
needle: value_to_domain(needle)?,
}),
FilterV1::All { filters } => filters
.iter()
.map(filter_to_domain)
.collect::<Result<_, _>>()
.map(Filter::All),
FilterV1::Any { filters } => filters
.iter()
.map(filter_to_domain)
.collect::<Result<_, _>>()
.map(Filter::Any),
FilterV1::Not { filter } => Ok(Filter::Not(Box::new(filter_to_domain(filter)?))),
}
}
fn sort_to_domain(field: &SortFieldV1) -> Result<SortField, WireValueError> {
Ok(SortField {
path: field_path(&field.path)?,
direction: match field.direction {
SortDirectionV1::Ascending => SortDirection::Ascending,
SortDirectionV1::Descending => SortDirection::Descending,
},
nulls: match field.nulls {
NullPlacementV1::First => NullPlacement::First,
NullPlacementV1::Last => NullPlacement::Last,
},
})
}
fn cursor_to_domain(cursor: &CursorV1) -> Result<Cursor, WireValueError> {
Ok(Cursor {
sort_values: cursor
.sort_values
.iter()
.map(|value| value.as_ref().map(value_to_domain).transpose())
.collect::<Result<_, _>>()?,
key: decode_key_hex(&cursor.key_hex)?,
})
}
fn aggregation_to_domain(plan: &AggregationPlanV1) -> Result<AggregationPlan, WireValueError> {
Ok(AggregationPlan {
group_by: plan
.group_by
.iter()
.map(|path| field_path(path))
.collect::<Result<_, _>>()?,
metrics: plan
.metrics
.iter()
.map(|metric| match metric {
NamedMetricV1::Count { name } => Ok(NamedMetric {
name: name.clone(),
metric: Metric::Count,
}),
NamedMetricV1::Sum { name, path } => Ok(NamedMetric {
name: name.clone(),
metric: Metric::Sum(field_path(path)?),
}),
NamedMetricV1::Min { name, path } => Ok(NamedMetric {
name: name.clone(),
metric: Metric::Min(field_path(path)?),
}),
NamedMetricV1::Max { name, path } => Ok(NamedMetric {
name: name.clone(),
metric: Metric::Max(field_path(path)?),
}),
})
.collect::<Result<_, _>>()?,
})
}
fn cursor_from_domain(cursor: &Cursor) -> CursorV1 {
CursorV1 {
sort_values: cursor
.sort_values
.iter()
.map(|value| value.as_ref().map(value_from_domain))
.collect(),
key_hex: encode_hex(&cursor.key),
}
}
fn aggregation_from_domain(result: &AggregationResult) -> AggregationResultV1 {
AggregationResultV1 {
grouped: result.grouped,
groups: result.groups.iter().map(group_from_domain).collect(),
}
}
fn group_from_domain(group: &GroupResult) -> GroupResultV1 {
GroupResultV1 {
key: group
.key
.iter()
.map(|value| match value {
None => GroupKeyValueV1::Missing,
Some(value) => GroupKeyValueV1::Value {
value: value_from_domain(value),
},
})
.collect(),
metrics: group.metrics.iter().map(metric_from_domain).collect(),
}
}
fn metric_from_domain(metric: &NamedMetricValue) -> NamedMetricResultV1 {
NamedMetricResultV1 {
name: metric.name.clone(),
result: match &metric.value {
MetricValue::Count(value) => MetricValueV1::Count { value: *value },
MetricValue::Integer(value) => MetricValueV1::Integer {
value: value.map(|value| value.to_string()),
},
MetricValue::Value(value) => MetricValueV1::Value {
value: value.as_ref().map(value_from_domain),
},
},
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use hyphae_query::{Filter, Value};
use super::{
BYTES_HEX_KEY, FilterV1, GroupKeyValueV1, GroupResultV1, QueryRequestV1, WireValueError,
value_from_domain, value_to_domain,
};
#[test]
fn natural_json_value_round_trips_domain_values() -> Result<(), WireValueError> {
let value = Value::Object(BTreeMap::from([
("bytes".to_owned(), Value::Bytes(vec![0, 255])),
("number".to_owned(), Value::Integer(-7)),
]));
assert_eq!(value_to_domain(&value_from_domain(&value))?, value);
assert_eq!(
value_to_domain(&serde_json::json!({BYTES_HEX_KEY: "00ff"}))?,
Value::Bytes(vec![0, 255])
);
Ok(())
}
#[test]
fn wire_query_rejects_noninteger_numbers_and_empty_paths() {
assert_eq!(
value_to_domain(&serde_json::json!(1.5)),
Err(WireValueError::NonIntegerNumber)
);
let request = QueryRequestV1 {
filter: FilterV1::Exists {
path: vec![String::new()],
},
sort: Vec::new(),
cursor: None,
limit: 1,
aggregation: None,
timeout_ms: None,
};
assert_eq!(request.to_domain(), Err(WireValueError::EmptyPathSegment));
assert_ne!(Filter::MatchAll, Filter::Any(Vec::new()));
}
#[test]
fn aggregation_group_keys_preserve_missing_versus_explicit_null()
-> Result<(), serde_json::Error> {
let missing = GroupResultV1 {
key: vec![GroupKeyValueV1::Missing],
metrics: Vec::new(),
};
let explicit_null = GroupResultV1 {
key: vec![GroupKeyValueV1::Value {
value: serde_json::Value::Null,
}],
metrics: Vec::new(),
};
assert_ne!(
serde_json::to_value(missing)?,
serde_json::to_value(explicit_null)?
);
Ok(())
}
}