luulam/AFFiNE
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat: impl text delta support (#14132)

Browse Source
This commit is contained in:
DarkSky
2025-12-22 03:16:16 +08:00
committed by GitHub
parent efbdee5508
commit 321965a424
31 changed files with 2299 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
packages/common/y-octo/core/src/doc/codec/any.rs
View File

@@ -529,18 +529,18 @@ impl Display for Any {
match self {
Self::True => write!(f, "true"),
Self::False => write!(f, "false"),
Self::String(s) => write!(f, "\"{}\"", s),
Self::Integer(i) => write!(f, "{}", i),
Self::Float32(v) => write!(f, "{}", v),
Self::Float64(v) => write!(f, "{}", v),
Self::BigInt64(v) => write!(f, "{}", v),
Self::String(s) => write!(f, "\"{s}\""),
Self::Integer(i) => write!(f, "{i}"),
Self::Float32(v) => write!(f, "{v}"),
Self::Float64(v) => write!(f, "{v}"),
Self::BigInt64(v) => write!(f, "{v}"),
Self::Object(map) => {
write!(f, "{{")?;
for (i, (key, value)) in map.iter().enumerate() {
if i > 0 {
write!(f, ", ")?;
}
write!(f, "{}: {}", key, value)?;
write!(f, "{key}: {value}")?;
}
write!(f, "}}")
}
@@ -550,11 +550,11 @@ impl Display for Any {
if i > 0 {
write!(f, ", ")?;
}
write!(f, "{}", value)?;
write!(f, "{value}")?;
}
write!(f, "]")
}
Self::Binary(buf) => write!(f, "{:?}", buf),
Self::Binary(buf) => write!(f, "{buf:?}"),
Self::Undefined => write!(f, "undefined"),
Self::Null => write!(f, "null"),
}

2
packages/common/y-octo/core/src/doc/codec/update.rs
View File

@@ -166,7 +166,7 @@ impl Update {
// merge two nodes, mark the index
merged_index.push(index + 1);
} else {
debug!("merge failed: {:?} {:?}", cur, next)
debug!("merge failed: {cur:?} {next:?}")
}
}

6
packages/common/y-octo/core/src/doc/publisher.rs
View File

@@ -96,13 +96,13 @@ impl DocPublisher {
let mut encoder = RawEncoder::default();
if let Err(e) = update.write(&mut encoder) {
warn!("Failed to encode document: {}", e);
warn!("Failed to encode document: {e}");
continue;
}
(encoder.into_inner(), history)
}
Err(e) => {
warn!("Failed to diff document: {}", e);
warn!("Failed to diff document: {e}");
continue;
}
};
@@ -117,7 +117,7 @@ impl DocPublisher {
cb(&binary, &history);
}))
.unwrap_or_else(|e| {
warn!("Failed to call subscriber: {:?}", e);
warn!("Failed to call subscriber: {e:?}");
});
}
} else {

8
packages/common/y-octo/core/src/doc/store.rs
View File

@@ -90,7 +90,7 @@ impl DocStore {
if let Some(last_struct) = structs.back() {
last_struct.clock() + last_struct.len()
} else {
warn!("client {} has no struct info", client);
warn!("client {client} has no struct info");
0
}
} else {
@@ -108,7 +108,7 @@ impl DocStore {
if let Some(last_struct) = structs.back() {
state.insert(*client, last_struct.clock() + last_struct.len());
} else {
warn!("client {} has no struct info", client);
warn!("client {client} has no struct info");
}
}
state
@@ -126,7 +126,7 @@ impl DocStore {
return Err(JwstCodecError::StructClockInvalid { expect, actually });
}
} else {
warn!("client {} has no struct info", client_id);
warn!("client {client_id} has no struct info");
}
structs.push_back(item);
}
@@ -593,7 +593,7 @@ impl DocStore {
self.delete_node(&Node::Item(item_owner_ref.clone()), Some(parent));
} else {
// adjust parent length
if this.parent_sub.is_none() {
if this.parent_sub.is_none() && this.countable() {
parent.len += this.len();
}
}

10
packages/common/y-octo/core/src/doc/types/list/mod.rs
View File

@@ -17,7 +17,7 @@ pub(crate) struct ItemPosition {
impl ItemPosition {
pub fn forward(&mut self) {
if let Some(right) = self.right.get() {
if !right.deleted() {
if right.indexable() {
self.index += right.len();
}
@@ -103,7 +103,7 @@ pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
while remaining > 0 {
if let Some(item) = pos.right.get() {
if !item.deleted() {
if item.indexable() {
let content_len = item.len();
if remaining < content_len {
pos.offset = remaining;
@@ -148,7 +148,9 @@ pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
content: Content,
) -> JwstCodecResult {
if let Some(markers) = &ty.markers {
markers.update_marker_changes(pos.index, content.clock_len() as i64);
if content.countable() {
markers.update_marker_changes(pos.index, content.clock_len() as i64);
}
}
let item = store.create_item(
@@ -214,7 +216,7 @@ pub(crate) trait ListType: AsInner<Inner = YTypeRef> {
while remaining > 0 {
if let Some(item) = pos.right.get() {
if !item.deleted() {
if item.indexable() {
let content_len = item.len();
if remaining < content_len {
store.split_node(item.id, remaining)?;

586
packages/common/y-octo/core/src/doc/types/text.rs
View File

@@ -1,12 +1,43 @@
use std::fmt::Display;
use std::{collections::BTreeMap, fmt::Display};
use super::list::ListType;
use crate::{impl_type, Content, JwstCodecResult};
use super::{list::ListType, AsInner};
use crate::{
doc::{DocStore, ItemRef, Node, Parent, Somr, YType, YTypeRef},
impl_type, Any, Content, JwstCodecError, JwstCodecResult,
};
impl_type!(Text);
impl ListType for Text {}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(untagged)]
pub enum TextInsert {
Text(String),
Embed(Vec<Any>),
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(untagged)]
pub enum TextDeltaOp {
Insert {
insert: TextInsert,
#[serde(skip_serializing_if = "Option::is_none")]
format: Option<TextAttributes>,
},
Retain {
retain: u64,
#[serde(skip_serializing_if = "Option::is_none")]
format: Option<TextAttributes>,
},
Delete {
delete: u64,
},
}
pub type TextDelta = Vec<TextDeltaOp>;
pub type TextAttributes = BTreeMap<String, Any>;
impl Text {
#[inline]
pub fn len(&self) -> u64 {
@@ -27,13 +58,113 @@ impl Text {
pub fn remove(&mut self, char_index: u64, len: u64) -> JwstCodecResult {
self.remove_at(char_index, len)
}
pub fn to_delta(&self) -> TextDelta {
let mut ops = Vec::new();
let mut attrs = TextAttributes::new();
for item_ref in self.iter_item() {
if let Some(item) = item_ref.get() {
match &item.content {
Content::Format { key, value } => {
if is_nullish(value) {
attrs.remove(key.as_str());
} else {
attrs.insert(key.to_string(), value.clone());
}
}
Content::String(text) => {
push_insert(&mut ops, TextInsert::Text(text.clone()), &attrs);
}
Content::Embed(embed) => {
push_insert(&mut ops, TextInsert::Embed(vec![embed.clone()]), &attrs);
}
Content::Any(any) => {
push_insert(&mut ops, TextInsert::Embed(any.clone()), &attrs);
}
Content::Json(values) => {
let converted = values
.iter()
.map(|value| {
value
.as_ref()
.map(|s| Any::String(s.clone()))
.unwrap_or(Any::Undefined)
})
.collect::<Vec<_>>();
push_insert(&mut ops, TextInsert::Embed(converted), &attrs);
}
Content::Binary(value) => {
push_insert(
&mut ops,
TextInsert::Embed(vec![Any::Binary(value.clone())]),
&attrs,
);
}
_ => {}
}
}
}
ops
}
pub fn apply_delta(&mut self, delta: &[TextDeltaOp]) -> JwstCodecResult {
let (mut store, mut ty) = self.as_inner().write().ok_or(JwstCodecError::DocReleased)?;
let parent = self.as_inner().clone();
let mut pos = TextPosition::new(parent, ty.start.clone());
for op in delta {
match op {
TextDeltaOp::Insert { insert, format } => {
let attrs = format.clone().unwrap_or_default();
match insert {
TextInsert::Text(text) => {
insert_text_content(
&mut store,
&mut ty,
&mut pos,
Content::String(text.clone()),
attrs,
)?;
}
TextInsert::Embed(values) => {
for value in values {
insert_text_content(
&mut store,
&mut ty,
&mut pos,
Content::Embed(value.clone()),
attrs.clone(),
)?;
}
}
}
}
TextDeltaOp::Retain { retain, format } => {
let attrs = format.clone().unwrap_or_default();
if attrs.is_empty() {
advance_text_position(&mut store, &mut pos, *retain)?;
} else {
format_text(&mut store, &mut ty, &mut pos, *retain, attrs)?;
}
}
TextDeltaOp::Delete { delete } => {
delete_text(&mut store, &mut ty, &mut pos, *delete)?;
}
}
}
Ok(())
}
}
impl Display for Text {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.iter_item().try_for_each(|item| {
if let Content::String(str) = &item.get().unwrap().content {
write!(f, "{}", str)
write!(f, "{str}")
} else {
Ok(())
}
@@ -50,15 +181,361 @@ impl serde::Serialize for Text {
}
}
struct TextPosition {
parent: YTypeRef,
left: ItemRef,
right: ItemRef,
index: u64,
attrs: TextAttributes,
}
impl TextPosition {
fn new(parent: YTypeRef, right: ItemRef) -> Self {
Self {
parent,
left: Somr::none(),
right,
index: 0,
attrs: TextAttributes::new(),
}
}
fn forward(&mut self) {
if let Some(right) = self.right.get() {
if !right.deleted() {
if let Content::Format { key, value } = &right.content {
if is_nullish(value) {
self.attrs.remove(key.as_str());
} else {
self.attrs.insert(key.to_string(), value.clone());
}
} else if right.countable() {
self.index += right.len();
}
}
self.left = self.right.clone();
self.right = right.right.clone();
}
}
}
fn is_nullish(value: &Any) -> bool {
matches!(value, Any::Null | Any::Undefined)
}
fn push_insert(ops: &mut Vec<TextDeltaOp>, insert: TextInsert, attrs: &TextAttributes) {
let format = if attrs.is_empty() {
None
} else {
Some(attrs.clone())
};
if let Some(TextDeltaOp::Insert {
insert: TextInsert::Text(prev),
format: prev_format,
}) = ops.last_mut()
{
if let TextInsert::Text(text) = insert {
if prev_format.as_ref() == format.as_ref() {
prev.push_str(&text);
return;
}
ops.push(TextDeltaOp::Insert {
insert: TextInsert::Text(text),
format,
});
return;
}
}
ops.push(TextDeltaOp::Insert { insert, format });
}
fn advance_text_position(
store: &mut DocStore,
pos: &mut TextPosition,
mut remaining: u64,
) -> JwstCodecResult {
while remaining > 0 {
let Some(item) = pos.right.get() else {
return Err(JwstCodecError::IndexOutOfBound(pos.index + remaining));
};
if item.deleted() {
pos.forward();
continue;
}
if matches!(item.content, Content::Format { .. }) {
pos.forward();
continue;
}
let item_len = item.len();
if remaining < item_len {
let (left, right) = store.split_node(item.id, remaining)?;
pos.left = left.as_item();
pos.right = right.as_item();
pos.index += remaining;
break;
}
remaining -= item_len;
pos.forward();
}
Ok(())
}
fn minimize_attribute_changes(pos: &mut TextPosition, attrs: &TextAttributes) {
loop {
let Some(item) = pos.right.get() else {
break;
};
if item.deleted() {
pos.forward();
continue;
}
if let Content::Format { key, value } = &item.content {
let attr = attrs.get(key.as_str()).cloned().unwrap_or(Any::Null);
if attr == *value {
pos.forward();
continue;
}
}
break;
}
}
fn insert_item(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
content: Content,
) -> JwstCodecResult {
if let Some(markers) = &ty.markers {
if content.countable() {
markers.update_marker_changes(pos.index, content.clock_len() as i64);
}
}
let item = store.create_item(
content,
pos.left.clone(),
pos.right.clone(),
Some(Parent::Type(pos.parent.clone())),
None,
);
let item_ref = item.clone();
store.integrate(Node::Item(item), 0, Some(ty))?;
pos.right = item_ref;
pos.forward();
Ok(())
}
fn insert_attributes(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
attrs: &TextAttributes,
) -> JwstCodecResult<TextAttributes> {
let mut negated = TextAttributes::new();
for (key, value) in attrs {
let current = pos.attrs.get(key.as_str()).cloned().unwrap_or(Any::Null);
if current == *value {
continue;
}
negated.insert(key.to_string(), current);
insert_item(
store,
ty,
pos,
Content::Format {
key: key.to_string(),
value: value.clone(),
},
)?;
}
Ok(negated)
}
fn insert_negated_attributes(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
mut negated: TextAttributes,
) -> JwstCodecResult {
loop {
let Some(item) = pos.right.get() else {
break;
};
if item.deleted() {
pos.forward();
continue;
}
if let Content::Format { key, value } = &item.content {
if let Some(negated_value) = negated.get(key.as_str()) {
if negated_value == value {
negated.remove(key.as_str());
pos.forward();
continue;
}
}
}
break;
}
for (key, value) in negated {
insert_item(
store,
ty,
pos,
Content::Format {
key: key.to_string(),
value,
},
)?;
}
Ok(())
}
fn insert_text_content(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
content: Content,
mut attrs: TextAttributes,
) -> JwstCodecResult {
for key in pos.attrs.keys() {
if !attrs.contains_key(key.as_str()) {
attrs.insert(key.to_string(), Any::Null);
}
}
minimize_attribute_changes(pos, &attrs);
let negated = insert_attributes(store, ty, pos, &attrs)?;
insert_item(store, ty, pos, content)?;
insert_negated_attributes(store, ty, pos, negated)?;
Ok(())
}
fn format_text(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
mut remaining: u64,
attrs: TextAttributes,
) -> JwstCodecResult {
if remaining == 0 {
return Ok(());
}
minimize_attribute_changes(pos, &attrs);
let mut negated = insert_attributes(store, ty, pos, &attrs)?;
while remaining > 0 {
let Some(item) = pos.right.get() else {
break;
};
if item.deleted() {
pos.forward();
continue;
}
match &item.content {
Content::Format { key, value } => {
if let Some(attr) = attrs.get(key.as_str()) {
if attr == value {
negated.remove(key.as_str());
} else {
negated.insert(key.to_string(), value.clone());
}
store.delete_item(item, Some(ty));
pos.forward();
} else {
pos.forward();
}
}
_ => {
let item_len = item.len();
if remaining < item_len {
store.split_node(item.id, remaining)?;
remaining = 0;
} else {
remaining -= item_len;
}
pos.forward();
}
}
}
insert_negated_attributes(store, ty, pos, negated)?;
Ok(())
}
fn delete_text(
store: &mut DocStore,
ty: &mut YType,
pos: &mut TextPosition,
mut remaining: u64,
) -> JwstCodecResult {
if remaining == 0 {
return Ok(());
}
let start = remaining;
while remaining > 0 {
let Some(item) = pos.right.get() else {
break;
};
if item.indexable() {
let item_len = item.len();
if remaining < item_len {
store.split_node(item.id, remaining)?;
remaining = 0;
} else {
remaining -= item_len;
}
store.delete_item(item, Some(ty));
}
pos.forward();
}
if let Some(markers) = &ty.markers {
markers.update_marker_changes(pos.index, -((start - remaining) as i64));
}
Ok(())
}
#[cfg(test)]
mod tests {
use rand::{Rng, SeedableRng};
use rand_chacha::ChaCha20Rng;
use yrs::{Options, Text, Transact};
use super::{TextAttributes, TextDeltaOp, TextInsert};
#[cfg(not(loom))]
use crate::sync::{Arc, AtomicUsize, Ordering};
use crate::{loom_model, sync::thread, Doc};
use crate::{loom_model, sync::thread, Any, Doc};
#[test]
fn test_manipulate_text() {
@@ -290,4 +767,103 @@ mod tests {
assert_eq!(text.to_string(), "hello great world!");
});
}
#[test]
fn test_text_delta_insert_format() {
loom_model!({
let doc = Doc::new();
let mut text = doc.get_or_create_text("text").unwrap();
let mut attrs = TextAttributes::new();
attrs.insert("bold".to_string(), Any::True);
text
.apply_delta(&[TextDeltaOp::Insert {
insert: TextInsert::Text("abc".to_string()),
format: Some(attrs.clone()),
}])
.unwrap();
assert_eq!(text.to_string(), "abc");
assert_eq!(
text.to_delta(),
vec![TextDeltaOp::Insert {
insert: TextInsert::Text("abc".to_string()),
format: Some(attrs),
}]
);
});
}
#[test]
fn test_text_delta_retain_format() {
loom_model!({
let doc = Doc::new();
let mut text = doc.get_or_create_text("text").unwrap();
text
.apply_delta(&[TextDeltaOp::Insert {
insert: TextInsert::Text("abc".to_string()),
format: None,
}])
.unwrap();
let mut attrs = TextAttributes::new();
attrs.insert("bold".to_string(), Any::True);
text
.apply_delta(&[TextDeltaOp::Retain {
retain: 1,
format: Some(attrs.clone()),
}])
.unwrap();
assert_eq!(
text.to_delta(),
vec![
TextDeltaOp::Insert {
insert: TextInsert::Text("a".to_string()),
format: Some(attrs),
},
TextDeltaOp::Insert {
insert: TextInsert::Text("bc".to_string()),
format: None,
}
]
);
});
}
#[test]
fn test_text_delta_utf16_retain() {
loom_model!({
let doc = Doc::new();
let mut text = doc.get_or_create_text("text").unwrap();
text
.apply_delta(&[TextDeltaOp::Insert {
insert: TextInsert::Text("😀".to_string()),
format: None,
}])
.unwrap();
let mut attrs = TextAttributes::new();
attrs.insert("bold".to_string(), Any::True);
text
.apply_delta(&[TextDeltaOp::Retain {
retain: 2,
format: Some(attrs.clone()),
}])
.unwrap();
assert_eq!(
text.to_delta(),
vec![TextDeltaOp::Insert {
insert: TextInsert::Text("😀".to_string()),
format: Some(attrs),
}]
);
});
}
}

4
packages/common/y-octo/core/src/doc/types/value.rs
View File

@@ -131,8 +131,8 @@ impl From<Doc> for Value {
impl Display for Value {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Value::Any(any) => write!(f, "{}", any),
Value::Text(text) => write!(f, "{}", text),
Value::Any(any) => write!(f, "{any}"),
Value::Text(text) => write!(f, "{text}"),
_ => write!(f, ""),
}
}

3
packages/common/y-octo/core/src/lib.rs
View File

@@ -9,7 +9,8 @@ pub use doc::{
encode_awareness_as_message, encode_update_as_message, merge_updates_v1, Any, Array, Awareness,
AwarenessEvent, Client, ClientMap, Clock, CrdtRead, CrdtReader, CrdtWrite, CrdtWriter, Doc,
DocOptions, HashMap as AHashMap, HashMapExt, History, HistoryOptions, Id, Map, RawDecoder,
RawEncoder, StateVector, StoreHistory, Text, Update, Value,
RawEncoder, StateVector, StoreHistory, Text, TextAttributes, TextDelta, TextDeltaOp, TextInsert,
Update, Value,
};
pub(crate) use doc::{Content, Item};
use log::{debug, warn};

2
packages/common/y-octo/core/src/protocol/sync.rs
View File

@@ -62,7 +62,7 @@ pub fn read_sync_message(input: &[u8]) -> IResult<&[u8], SyncMessage> {
let (awareness_tail, awareness) = read_awareness(update)?;
let tail_len = awareness_tail.len();
if tail_len > 0 {
debug!("awareness update has trailing bytes: {}", tail_len);
debug!("awareness update has trailing bytes: {tail_len}");
debug_assert!(tail_len > 0, "awareness update has trailing bytes");
}
awareness