use cloudillo_types::crdt_adapter::CrdtUpdate;
use serde_json::Value;
use yrs::{
Any, ArrayRef, Doc, GetString, Map, MapRef, OffsetKind, Options, Out, ReadTxn, TextRef,
Transact, Update, branch::BranchPtr, types::ToJson, updates::decoder::Decode,
};
use crate::prelude::*;
pub async fn export_all(app: &App, tn_id: TnId, doc_id: &str) -> ClResult<Vec<(Box<str>, Value)>> {
let updates = app.crdt_adapter.get_updates(tn_id, doc_id).await?;
if updates.is_empty() {
return Ok(Vec::new());
}
let owned_id = doc_id.to_owned();
app.worker
.run_slow(move || materialize(&updates, &owned_id))
.await
.map_err(|e| Error::Internal(format!("Worker pool failed reading CRDT doc: {e}")))
}
fn materialize(updates: &[CrdtUpdate], doc_id: &str) -> Vec<(Box<str>, Value)> {
let doc = Doc::with_options(Options { offset_kind: OffsetKind::Utf16, ..Default::default() });
{
let mut txn = doc.transact_mut();
for (idx, stored) in updates.iter().enumerate() {
match Update::decode_v1(&stored.data) {
Ok(update) => {
if let Err(e) = txn.apply_update(update) {
warn!(doc_id, idx, error = %e, "CRDT update failed to apply while indexing");
}
}
Err(e) => {
warn!(doc_id, idx, error = %e, "CRDT update failed to decode while indexing");
}
}
}
}
let txn = doc.transact();
let mut out = Vec::new();
let mut roots: Vec<(&str, Out)> = txn.root_refs().collect();
roots.sort_unstable_by(|a, b| a.0.cmp(b.0));
for (root, value) in roots {
if matches!(value, Out::YMap(_) | Out::YArray(_) | Out::YText(_) | Out::UndefinedRef(_)) {
collect_root(&txn, doc_id, root, &value, &mut out);
}
}
out
}
const TEXT_ENTRY: &str = "_";
const TEXT_HEADING_MAX_CHARS: usize = 120;
fn first_line(text: &str) -> String {
let line = text.lines().map(str::trim).find(|l| !l.is_empty()).unwrap_or_default();
line.chars().take(TEXT_HEADING_MAX_CHARS).collect()
}
fn collect_root<T: ReadTxn>(
txn: &T,
doc_id: &str,
root: &str,
value: &Out,
out: &mut Vec<(Box<str>, Value)>,
) {
let Some(ptr) = value.try_branch().map(BranchPtr::from) else { return };
let mut truncated = false;
let map = MapRef::from(ptr);
if map.len(txn) > 0 {
let mut entries: Vec<(&str, Out)> = map.iter(txn).collect();
entries.sort_unstable_by(|a, b| a.0.cmp(b.0));
for (key, entry) in entries {
let json = any_to_json(&entry.to_json(txn), MAX_ANY_DEPTH, &mut truncated);
out.push((format!("{root}/{key}").into(), json));
}
if truncated {
warn!(doc_id, root, MAX_ANY_DEPTH, "CRDT root nested past the indexing depth limit");
}
return;
}
let text = TextRef::from(ptr).get_string(txn);
if !text.is_empty() {
if !text.trim().is_empty() {
let heading = first_line(&text);
out.push((
format!("{root}/{TEXT_ENTRY}").into(),
serde_json::json!({ "t": text, "h": heading }),
));
}
return;
}
let read = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
any_to_json(&ArrayRef::from(ptr).to_json(txn), MAX_ANY_DEPTH, &mut truncated)
}));
let Ok(json) = read else {
warn!(root, "CRDT root could not be read as a sequence; skipping");
return;
};
if truncated {
warn!(doc_id, root, MAX_ANY_DEPTH, "CRDT root nested past the indexing depth limit");
}
let Value::Array(items) = json else { return };
if items.iter().any(Value::is_string) {
let text: String = items.iter().filter_map(Value::as_str).collect();
if !text.trim().is_empty() {
let heading = first_line(&text);
out.push((
format!("{root}/{TEXT_ENTRY}").into(),
serde_json::json!({ "t": text, "h": heading }),
));
}
return;
}
if items.iter().all(|i| !i.is_object() && !i.is_array()) {
return;
}
for (i, item) in items.into_iter().enumerate() {
out.push((format!("{root}/{i}").into(), item));
}
}
const MAX_ANY_DEPTH: usize = 32;
fn any_to_json(any: &Any, depth: usize, truncated: &mut bool) -> Value {
match any {
Any::Null | Any::Undefined | Any::Buffer(_) => Value::Null,
Any::Bool(b) => Value::Bool(*b),
Any::Number(n) => serde_json::Number::from_f64(*n).map_or(Value::Null, Value::Number),
Any::BigInt(i) => Value::Number((*i).into()),
Any::String(s) => Value::String(s.to_string()),
Any::Array(_) | Any::Map(_) if depth == 0 => {
*truncated = true;
Value::Null
}
Any::Array(items) => {
Value::Array(items.iter().map(|i| any_to_json(i, depth - 1, truncated)).collect())
}
Any::Map(map) => {
let mut entries: Vec<(&String, &Any)> = map.iter().collect();
entries.sort_unstable_by(|a, b| a.0.cmp(b.0));
Value::Object(
entries
.into_iter()
.map(|(k, v)| (k.clone(), any_to_json(v, depth - 1, truncated)))
.collect(),
)
}
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use yrs::{Array, ArrayPrelim, MapPrelim, Text};
use super::*;
fn updates_of(doc: &Doc) -> Vec<CrdtUpdate> {
let data = doc.transact().encode_state_as_update_v1(&yrs::StateVector::default());
vec![CrdtUpdate::with_client(data, "test".to_owned())]
}
#[test]
fn a_root_map_becomes_one_document_per_key() {
let doc = Doc::new();
let pages = doc.get_or_insert_map("p");
{
let mut txn = doc.transact_mut();
pages.insert(&mut txn, "page1", MapPrelim::from([("ti", "Bevezetés")]));
pages.insert(&mut txn, "page2", MapPrelim::from([("ti", "Részletek")]));
}
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 2);
assert_eq!(&*out[0].0, "p/page1");
assert_eq!(out[0].1["ti"], serde_json::json!("Bevezetés"));
assert_eq!(&*out[1].0, "p/page2");
}
#[test]
fn a_root_array_becomes_one_document_per_index() {
let doc = Doc::new();
let slides = doc.get_or_insert_array("s");
{
let mut txn = doc.transact_mut();
slides.push_back(&mut txn, MapPrelim::from([("ti", "First")]));
slides.push_back(&mut txn, MapPrelim::from([("ti", "Second")]));
}
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 2);
assert_eq!(&*out[0].0, "s/0");
assert_eq!(out[0].1["ti"], serde_json::json!("First"));
assert_eq!(out[1].1["ti"], serde_json::json!("Second"));
}
#[test]
fn roots_come_out_in_name_order() {
let doc = Doc::new();
let second = doc.get_or_insert_map("z");
let first = doc.get_or_insert_map("a");
{
let mut txn = doc.transact_mut();
second.insert(&mut txn, "k", MapPrelim::from([("ti", "Utolsó")]));
first.insert(&mut txn, "k", MapPrelim::from([("ti", "Első")]));
}
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 2);
assert_eq!(&*out[0].0, "a/k");
assert_eq!(&*out[1].0, "z/k");
}
#[test]
fn nested_structures_survive_the_conversion() {
let doc = Doc::new();
let blocks = doc.get_or_insert_map("b");
{
let mut txn = doc.transact_mut();
blocks.insert(
&mut txn,
"blk",
MapPrelim::from([("c", ArrayPrelim::from(["hello", "world"]))]),
);
}
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 1);
assert_eq!(out[0].1["c"], serde_json::json!(["hello", "world"]));
}
#[test]
fn a_text_root_becomes_one_entry_holding_the_whole_stream() {
let doc = Doc::new();
let text = doc.get_or_insert_text("body");
{
let mut txn = doc.transact_mut();
text.push(&mut txn, "A címsor\nés a törzsszöveg.");
}
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 1);
assert_eq!(&*out[0].0, "body/_");
assert_eq!(out[0].1["t"], serde_json::json!("A címsor\nés a törzsszöveg."));
assert_eq!(out[0].1["h"], serde_json::json!("A címsor"), "the heading is the first line");
}
#[test]
fn a_blank_text_root_yields_nothing() {
let doc = Doc::new();
let text = doc.get_or_insert_text("body");
{
let mut txn = doc.transact_mut();
text.push(&mut txn, " \n\n");
}
assert!(materialize(&updates_of(&doc), "f1~doc").is_empty());
}
#[test]
fn a_text_root_with_an_embed_is_still_one_entry() {
let doc = Doc::new();
let text = doc.get_or_insert_text("body");
{
let mut txn = doc.transact_mut();
text.push(&mut txn, "before ");
text.insert_embed(&mut txn, 7, MapPrelim::from([("image", "pic.png")]));
text.insert(&mut txn, 8, " after");
}
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 1, "an embed must not split the document into positional rows");
assert_eq!(&*out[0].0, "body/_");
assert_eq!(out[0].1["t"], serde_json::json!("before after"));
}
#[test]
fn a_multi_chunk_non_ascii_text_root_does_not_hang() {
let doc =
Doc::with_options(Options { offset_kind: OffsetKind::Utf16, ..Default::default() });
let text = doc.get_or_insert_text("body");
{
let mut txn = doc.transact_mut();
text.push(&mut txn, "árvíztűrő tükörfúrógép");
text.remove_range(&mut txn, 9, 1); }
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 1);
assert_eq!(&*out[0].0, "body/_");
assert_eq!(out[0].1["t"], serde_json::json!("árvíztűrőtükörfúrógép"));
}
#[test]
fn a_blank_multi_chunk_non_ascii_text_root_does_not_hang() {
let doc =
Doc::with_options(Options { offset_kind: OffsetKind::Utf16, ..Default::default() });
let text = doc.get_or_insert_text("body");
{
let mut txn = doc.transact_mut();
text.push(&mut txn, "\u{3000}\u{3000}\u{00a0}\u{3000}");
text.remove_range(&mut txn, 1, 1); }
let updates = updates_of(&doc);
let (tx, rx) = std::sync::mpsc::channel();
let worker = std::thread::spawn(move || {
let _ = tx.send(materialize(&updates, "f1~doc"));
});
let out = rx
.recv_timeout(std::time::Duration::from_secs(10))
.expect("materialize spun on a blank multi-chunk non-ASCII text root");
worker.join().expect("materialize thread panicked");
assert!(out.is_empty(), "a blank text root carries nothing worth indexing");
}
#[test]
fn a_root_array_of_plain_strings_is_still_one_text_entry() {
let doc = Doc::new();
let lines = doc.get_or_insert_array("l");
{
let mut txn = doc.transact_mut();
lines.push_back(&mut txn, "Első sor");
lines.push_back(&mut txn, " és a többi");
}
let out = materialize(&updates_of(&doc), "f1~doc");
assert_eq!(out.len(), 1);
assert_eq!(&*out[0].0, "l/_");
assert_eq!(out[0].1["t"], serde_json::json!("Első sor és a többi"));
}
#[test]
fn a_root_array_of_loose_scalars_is_still_skipped() {
let doc = Doc::new();
let nums = doc.get_or_insert_array("n");
{
let mut txn = doc.transact_mut();
nums.push_back(&mut txn, Any::Number(1.0));
nums.push_back(&mut txn, Any::Bool(true));
}
assert!(materialize(&updates_of(&doc), "f1~doc").is_empty());
}
#[test]
fn a_corrupt_update_does_not_lose_the_rest_of_the_log() {
let doc = Doc::new();
let pages = doc.get_or_insert_map("p");
{
let mut txn = doc.transact_mut();
pages.insert(&mut txn, "page1", MapPrelim::from([("ti", "Kept")]));
}
let mut updates = updates_of(&doc);
updates.insert(0, CrdtUpdate::with_client(vec![0xff, 0xff, 0xff], "test".to_owned()));
let out = materialize(&updates, "f1~doc");
assert_eq!(out.len(), 1, "the readable update must still be indexed");
assert_eq!(out[0].1["ti"], serde_json::json!("Kept"));
}
#[test]
fn an_empty_log_yields_nothing() {
assert!(materialize(&[], "f1~doc").is_empty());
}
fn to_json(any: &Any) -> Value {
let mut truncated = false;
any_to_json(any, MAX_ANY_DEPTH, &mut truncated)
}
#[test]
fn binary_and_non_finite_values_become_null_rather_than_noise() {
assert_eq!(to_json(&Any::Buffer(Arc::from([1u8, 2, 3]))), Value::Null);
assert_eq!(to_json(&Any::Number(f64::NAN)), Value::Null);
assert_eq!(to_json(&Any::BigInt(-7)), serde_json::json!(-7));
}
#[test]
fn nesting_past_the_depth_limit_is_clipped_rather_than_followed() {
let mut any = Any::String("deep".into());
for _ in 0..(MAX_ANY_DEPTH + 8) {
any = Any::Array(Arc::from([any]));
}
let mut truncated = false;
let mut value = any_to_json(&any, MAX_ANY_DEPTH, &mut truncated);
assert!(truncated, "clipping must be observable to the caller");
for _ in 0..MAX_ANY_DEPTH {
let Value::Array(items) = value else { panic!("expected an array level") };
value = items.into_iter().next().expect("each level holds one child");
}
assert_eq!(value, Value::Null);
}
}