use crate::error::{DbError, DbResult};
use crate::storage::document::Document;
use serde::{Deserialize, Serialize};
pub const DOC_FORMAT_VERSION: u8 = 2;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DocumentWithVersion {
pub version: u8,
#[serde(rename = "_key")]
pub key: String,
#[serde(rename = "_id")]
pub id: String,
#[serde(rename = "_rev")]
pub rev: String,
#[serde(rename = "_created_at")]
pub created_at: chrono::DateTime<chrono::Utc>,
#[serde(rename = "_updated_at")]
pub updated_at: chrono::DateTime<chrono::Utc>,
#[serde(with = "serde_bytes")]
pub data: Vec<u8>,
}
impl DocumentWithVersion {
pub fn from_doc(doc: &Document) -> Self {
let data_bytes = rmp_serde::to_vec(&doc.data).unwrap_or_default();
Self {
version: DOC_FORMAT_VERSION,
key: doc.key.clone(),
id: doc.id.clone(),
rev: doc.rev.clone(),
created_at: doc.created_at,
updated_at: doc.updated_at,
data: data_bytes,
}
}
pub fn to_doc(&self) -> Document {
let data: serde_json::Value = rmp_serde::from_slice(&self.data)
.or_else(|_| serde_json::from_slice(&self.data))
.unwrap_or_default();
Document {
key: self.key.clone(),
id: self.id.clone(),
rev: self.rev.clone(),
created_at: self.created_at,
updated_at: self.updated_at,
data,
}
}
pub fn into_doc(self) -> Document {
let data: serde_json::Value = rmp_serde::from_slice(&self.data)
.or_else(|_| serde_json::from_slice(&self.data))
.unwrap_or_default();
Document {
key: self.key,
id: self.id,
rev: self.rev,
created_at: self.created_at,
updated_at: self.updated_at,
data,
}
}
}
const KEY_FIELD: &str = "_key";
const ID_FIELD: &str = "_id";
const REV_FIELD: &str = "_rev";
const CREATED_AT_FIELD: &str = "_created_at";
const UPDATED_AT_FIELD: &str = "_updated_at";
#[derive(Deserialize)]
struct StoredDocRef<'a> {
#[allow(dead_code)]
version: u8,
key: &'a str,
id: &'a str,
rev: &'a str,
created_at: &'a str,
updated_at: &'a str,
#[serde(with = "serde_bytes")]
data: &'a [u8],
}
fn stored_timestamp_to_rfc3339(stored: &str) -> String {
if let Some(base) = stored.strip_suffix('Z') {
let mut out = String::with_capacity(base.len() + 6);
out.push_str(base);
out.push_str("+00:00");
return out;
}
chrono::DateTime::parse_from_rfc3339(stored)
.map(|dt| dt.with_timezone(&chrono::Utc).to_rfc3339())
.unwrap_or_else(|_| stored.to_string())
}
struct MapKey<'de>(std::borrow::Cow<'de, str>);
impl<'de> Deserialize<'de> for MapKey<'de> {
fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
struct KeyVisitor;
impl<'de> serde::de::Visitor<'de> for KeyVisitor {
type Value = MapKey<'de>;
fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.write_str("a string key")
}
fn visit_borrowed_str<E>(self, v: &'de str) -> Result<Self::Value, E> {
Ok(MapKey(std::borrow::Cow::Borrowed(v)))
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E> {
Ok(MapKey(std::borrow::Cow::Owned(v.to_owned())))
}
fn visit_string<E>(self, v: String) -> Result<Self::Value, E> {
Ok(MapKey(std::borrow::Cow::Owned(v)))
}
}
d.deserialize_str(KeyVisitor)
}
}
struct ProjectedMap<'f> {
fields: &'f [String],
}
impl<'de> serde::de::DeserializeSeed<'de> for ProjectedMap<'_> {
type Value = serde_json::Map<String, serde_json::Value>;
fn deserialize<D: serde::Deserializer<'de>>(self, d: D) -> Result<Self::Value, D::Error> {
d.deserialize_map(self)
}
}
impl<'de> serde::de::Visitor<'de> for ProjectedMap<'_> {
type Value = serde_json::Map<String, serde_json::Value>;
fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.write_str("a document map")
}
fn visit_map<A: serde::de::MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
let mut out = serde_json::Map::new();
while let Some(MapKey(key)) = map.next_key()? {
if self.fields.iter().any(|f| f.as_str() == key.as_ref()) {
out.insert(key.into_owned(), map.next_value()?);
} else {
map.next_value::<serde::de::IgnoredAny>()?;
}
}
Ok(out)
}
}
pub fn deserialize_doc_projected(bytes: &[u8], fields: &[String]) -> DbResult<serde_json::Value> {
if bytes.first() == Some(&2) {
if let Ok(stored) = bincode::deserialize::<StoredDocRef>(&bytes[1..]) {
let mut de = rmp_serde::Deserializer::from_read_ref(stored.data);
if let Ok(mut map) =
serde::de::DeserializeSeed::deserialize(ProjectedMap { fields }, &mut de)
{
for field in fields {
let value = match field.as_str() {
KEY_FIELD => stored.key.into(),
ID_FIELD => stored.id.into(),
REV_FIELD => stored.rev.into(),
CREATED_AT_FIELD => stored_timestamp_to_rfc3339(stored.created_at).into(),
UPDATED_AT_FIELD => stored_timestamp_to_rfc3339(stored.updated_at).into(),
_ => continue,
};
map.insert(field.clone(), value);
}
return Ok(serde_json::Value::Object(map));
}
}
}
let value = deserialize_doc_as_value(bytes)?;
Ok(match value {
serde_json::Value::Object(mut map) => {
map.retain(|k, _| fields.iter().any(|f| f == k));
serde_json::Value::Object(map)
}
other => other,
})
}
pub fn deserialize_doc_as_value(bytes: &[u8]) -> DbResult<serde_json::Value> {
if bytes.is_empty() {
return Err(DbError::DocumentNotFound("empty bytes".to_string()));
}
match bytes[0] {
2 => {
if let Ok(stored) = bincode::deserialize::<StoredDocRef>(&bytes[1..]) {
let data: serde_json::Value =
rmp_serde::from_slice(stored.data).unwrap_or_default();
if let serde_json::Value::Object(mut map) = data {
map.insert(KEY_FIELD.to_owned(), stored.key.into());
map.insert(ID_FIELD.to_owned(), stored.id.into());
map.insert(REV_FIELD.to_owned(), stored.rev.into());
map.insert(
CREATED_AT_FIELD.to_owned(),
stored_timestamp_to_rfc3339(stored.created_at).into(),
);
map.insert(
UPDATED_AT_FIELD.to_owned(),
stored_timestamp_to_rfc3339(stored.updated_at).into(),
);
return Ok(serde_json::Value::Object(map));
}
}
let dwv: DocumentWithVersion = bincode::deserialize(&bytes[1..])
.map_err(|e| DbError::InternalError(format!("Deserialization failed: {}", e)))?;
let data: serde_json::Value = rmp_serde::from_slice(&dwv.data).unwrap_or_default();
if let serde_json::Value::Object(mut map) = data {
map.insert(KEY_FIELD.to_owned(), serde_json::Value::String(dwv.key));
map.insert(ID_FIELD.to_owned(), serde_json::Value::String(dwv.id));
map.insert(REV_FIELD.to_owned(), serde_json::Value::String(dwv.rev));
map.insert(
CREATED_AT_FIELD.to_owned(),
serde_json::Value::String(dwv.created_at.to_rfc3339()),
);
map.insert(
UPDATED_AT_FIELD.to_owned(),
serde_json::Value::String(dwv.updated_at.to_rfc3339()),
);
Ok(serde_json::Value::Object(map))
} else {
Ok(DocumentWithVersion {
version: 2,
key: dwv.key,
id: dwv.id,
rev: dwv.rev,
created_at: dwv.created_at,
updated_at: dwv.updated_at,
data: Vec::new(),
}
.into_doc()
.into_value())
}
}
_ => {
deserialize_doc(bytes).map(|doc| doc.into_value())
}
}
}
pub fn serialize_doc(doc: &Document) -> DbResult<Vec<u8>> {
let doc_with_version = DocumentWithVersion::from_doc(doc);
let mut bytes = Vec::new();
bytes.push(DOC_FORMAT_VERSION);
bincode::serialize_into(&mut bytes, &doc_with_version)
.map_err(|e| DbError::InternalError(format!("Serialization failed: {}", e)))?;
Ok(bytes)
}
pub fn deserialize_doc(bytes: &[u8]) -> DbResult<Document> {
if bytes.is_empty() {
return Err(DbError::DocumentNotFound("empty bytes".to_string()));
}
match bytes[0] {
2 => bincode::deserialize::<DocumentWithVersion>(&bytes[1..])
.map_err(|e| DbError::InternalError(format!("Deserialization failed: {}", e)))
.map(|doc_with_version| doc_with_version.into_doc()),
1 => {
bincode::deserialize::<DocumentWithVersion>(&bytes[1..])
.map_err(|e| {
DbError::InternalError(format!("Legacy v1 deserialization failed: {}", e))
})
.map(|doc_with_version| {
to_doc_legacy_v1(&doc_with_version)
})
}
_ => {
let doc: Document = serde_json::from_slice(bytes).map_err(|e| {
DbError::InternalError(format!("Legacy JSON deserialization failed: {}", e))
})?;
Ok(doc)
}
}
}
fn to_doc_legacy_v1(doc_with_version: &DocumentWithVersion) -> Document {
let data: serde_json::Value =
serde_json::from_slice(&doc_with_version.data).unwrap_or_default();
Document {
key: doc_with_version.key.clone(),
id: doc_with_version.id.clone(),
rev: doc_with_version.rev.clone(),
created_at: doc_with_version.created_at,
updated_at: doc_with_version.updated_at,
data,
}
}
pub fn serialize_to_json(doc: &Document) -> DbResult<Vec<u8>> {
serde_json::to_vec(doc)
.map_err(|e| DbError::InternalError(format!("JSON serialization failed: {}", e)))
}
pub fn deserialize_from_json(bytes: &[u8]) -> DbResult<Document> {
serde_json::from_slice(bytes)
.map_err(|e| DbError::InternalError(format!("JSON deserialization failed: {}", e)))
}
pub fn needs_migration(bytes: &[u8]) -> bool {
!bytes.is_empty() && bytes[0] != DOC_FORMAT_VERSION
}
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::document::Document;
use serde_json::json;
fn create_test_doc() -> Document {
Document::with_key(
"test_collection",
"test-key-123".to_string(),
json!({
"name": "Alice",
"age": 30,
"active": true,
"score": 98.5,
"tags": ["user", "premium"],
"metadata": {
"created_by": "admin",
"version": 1
}
}),
)
}
#[test]
fn test_serialize_deserialize_roundtrip() {
let doc = create_test_doc();
let bytes = serialize_doc(&doc).unwrap();
let deserialized = deserialize_doc(&bytes).unwrap();
assert_eq!(doc.key, deserialized.key);
assert_eq!(doc.id, deserialized.id);
assert_eq!(doc.rev, deserialized.rev);
assert_eq!(doc.data, deserialized.data);
}
#[test]
fn test_serialization_size() {
let doc = create_test_doc();
let json_bytes = serde_json::to_vec(&doc).unwrap();
let optimized_bytes = serialize_doc(&doc).unwrap();
let data_json_bytes = serde_json::to_vec(&doc.data).unwrap();
let data_msgpack_bytes = rmp_serde::to_vec(&doc.data).unwrap();
println!("Full JSON size: {} bytes", json_bytes.len());
println!("Optimized size: {} bytes", optimized_bytes.len());
println!("Data field JSON size: {} bytes", data_json_bytes.len());
println!(
"Data field MessagePack size: {} bytes",
data_msgpack_bytes.len()
);
println!(
"MessagePack data reduction: {:.1}%",
100.0 * (1.0 - data_msgpack_bytes.len() as f64 / data_json_bytes.len() as f64)
);
println!(
"Overall size reduction: {:.1}%",
100.0 * (1.0 - optimized_bytes.len() as f64 / json_bytes.len() as f64)
);
assert!(
data_msgpack_bytes.len() < data_json_bytes.len(),
"MessagePack should be smaller than JSON for data field"
);
}
#[test]
fn test_legacy_json_migration() {
let doc = create_test_doc();
let json_bytes = serde_json::to_vec(&doc).unwrap();
assert!(needs_migration(&json_bytes));
let deserialized = deserialize_doc(&json_bytes).unwrap();
assert_eq!(doc.key, deserialized.key);
assert_eq!(doc.data, deserialized.data);
}
#[test]
fn test_legacy_v1_format_compatibility() {
let doc = create_test_doc();
let data_bytes = serde_json::to_vec(&doc.data).unwrap();
let doc_v1 = DocumentWithVersion {
version: 1, key: doc.key.clone(),
id: doc.id.clone(),
rev: doc.rev.clone(),
created_at: doc.created_at,
updated_at: doc.updated_at,
data: data_bytes,
};
let mut bytes_v1 = vec![1u8]; bincode::serialize_into(&mut bytes_v1, &doc_v1).unwrap();
assert!(needs_migration(&bytes_v1));
let deserialized = deserialize_doc(&bytes_v1).unwrap();
assert_eq!(doc.key, deserialized.key);
assert_eq!(doc.data, deserialized.data);
assert_eq!(doc.id, deserialized.id);
}
#[test]
fn test_current_format_no_migration() {
let doc = create_test_doc();
let bincode_bytes = serialize_doc(&doc).unwrap();
assert!(!needs_migration(&bincode_bytes));
}
#[test]
fn test_empty_bytes_error() {
let result = deserialize_doc(&[]);
assert!(result.is_err());
if let Err(DbError::DocumentNotFound(msg)) = result {
assert_eq!(msg, "empty bytes");
} else {
panic!("Expected DocumentNotFound error");
}
}
#[test]
fn test_complex_document_roundtrip() {
let complex_data = json!({
"nested": {
"deeply": {
"nested": {
"value": "string",
"number": 42,
"float": 3.5,
"bool": false,
"null": null
}
}
},
"array_of_objects": [
{"id": 1, "name": "first"},
{"id": 2, "name": "second"},
{"id": 3, "name": "third"}
],
"mixed_array": [1, "two", 3.0, true, null]
});
let doc = Document::with_key("test", "complex".to_string(), complex_data);
let bytes = serialize_doc(&doc).unwrap();
let deserialized = deserialize_doc(&bytes).unwrap();
assert_eq!(doc.key, deserialized.key);
assert_eq!(doc.data, deserialized.data);
}
#[test]
fn test_special_characters() {
let doc = Document::with_key(
"test",
"special".to_string(),
json!({
"unicode": "Hello 世界 🌍 Café",
"quotes": "He said \"Hello\"",
"newlines": "Line1\nLine2\r\nLine3",
"tabs": "Col1\tCol2"
}),
);
let bytes = serialize_doc(&doc).unwrap();
let deserialized = deserialize_doc(&bytes).unwrap();
assert_eq!(doc.data, deserialized.data);
}
#[test]
fn fast_value_path_matches_document_path() {
use chrono::{TimeZone, Timelike};
for nanos in [0u32, 5_000_000, 123_456_000, 123_456_789, 1] {
let mut doc = create_test_doc();
let ts = chrono::Utc
.with_ymd_and_hms(2026, 9, 28, 8, 15, 6)
.unwrap()
.with_nanosecond(nanos)
.unwrap();
doc.created_at = ts;
doc.updated_at = ts + chrono::Duration::seconds(1);
let bytes = serialize_doc(&doc).unwrap();
let fast = deserialize_doc_as_value(&bytes).unwrap();
let slow = deserialize_doc(&bytes).unwrap().into_value();
assert_eq!(fast, slow, "nanos = {nanos}");
assert_eq!(fast["_created_at"], ts.to_rfc3339());
}
}
#[test]
fn projected_decode_matches_full_decode() {
let doc = create_test_doc();
let bytes = serialize_doc(&doc).unwrap();
let full = deserialize_doc_as_value(&bytes).unwrap();
let fields: Vec<String> = ["name", "tags", "_key", "_created_at", "absent"]
.iter()
.map(|s| s.to_string())
.collect();
let projected = deserialize_doc_projected(&bytes, &fields).unwrap();
let obj = projected.as_object().unwrap();
assert_eq!(obj.len(), 4, "{projected}");
for f in ["name", "tags", "_key", "_created_at"] {
assert_eq!(obj[f], full[f], "{f}");
}
assert!(obj.get("absent").is_none());
}
}