···8383pub fn CollabCoordinator(props: CollabCoordinatorProps) -> Element {
8484 #[cfg(target_arch = "wasm32")]
8585 {
8686- use super::worker::{WorkerInput, WorkerOutput};
8786 use crate::collab_context::CollabDebugState;
8887 use crate::fetch::Fetcher;
8988 use futures_util::stream::SplitSink;
···9291 use gloo_worker::reactor::ReactorBridge;
9392 use jacquard::IntoStatic;
9493 use weaver_common::WeaverExt;
9595-9696- use super::worker::EditorReactor;
9494+ use weaver_editor_crdt::{EditorReactor, WorkerInput, WorkerOutput};
97959896 let fetcher = use_context::<Fetcher>();
9997
···11-//! Web Worker for offloading expensive editor operations.
11+//! Web Worker reactor for offloading expensive editor operations.
22//!
33//! This worker maintains a shadow copy of the Loro document and handles
44//! CPU-intensive operations like snapshot export and base64 encoding
55//! off the main thread.
66//!
77-//! When the `collab-worker` feature is enabled, also handles iroh P2P
77+//! When the `collab` feature is enabled, also handles iroh P2P
88//! networking for real-time collaboration.
991010#[cfg(all(target_family = "wasm", target_os = "unknown"))]
···138138 use gloo_worker::reactor::{ReactorScope, reactor};
139139 use weaver_common::transport::CollaboratorInfo;
140140141141- #[cfg(feature = "collab-worker")]
141141+ #[cfg(feature = "collab")]
142142 use jacquard::smol_str::ToSmolStr;
143143- #[cfg(feature = "collab-worker")]
143143+ #[cfg(feature = "collab")]
144144 use std::sync::Arc;
145145- #[cfg(feature = "collab-worker")]
145145+ #[cfg(feature = "collab")]
146146 use weaver_common::transport::{
147147 CollabMessage, CollabNode, CollabSession, PresenceTracker, SessionEvent, TopicId,
148148 parse_node_id,
149149 };
150150151151 /// Internal event from gossip handler task to main reactor loop.
152152- #[cfg(feature = "collab-worker")]
152152+ #[cfg(feature = "collab")]
153153 enum CollabEvent {
154154 RemoteUpdates { data: Vec<u8> },
155155 PresenceChanged(PresenceSnapshot),
···162162 let mut doc: Option<loro::LoroDoc> = None;
163163 let mut draft_key = SmolStr::default();
164164165165- // Collab state (only used when collab-worker feature enabled)
166166- #[cfg(feature = "collab-worker")]
165165+ // Collab state (only used when collab feature enabled)
166166+ #[cfg(feature = "collab")]
167167 let mut collab_node: Option<Arc<CollabNode>> = None;
168168- #[cfg(feature = "collab-worker")]
168168+ #[cfg(feature = "collab")]
169169 let mut collab_session: Option<Arc<CollabSession>> = None;
170170- #[cfg(feature = "collab-worker")]
170170+ #[cfg(feature = "collab")]
171171 let mut collab_event_rx: Option<tokio::sync::mpsc::UnboundedReceiver<CollabEvent>> = None;
172172- #[cfg(feature = "collab-worker")]
172172+ #[cfg(feature = "collab")]
173173 const OUR_COLOR: u32 = 0x4ECDC4FF;
174174175175 // Helper enum for racing coordinator messages vs collab events
176176- #[cfg(feature = "collab-worker")]
176176+ #[cfg(feature = "collab")]
177177 enum RaceResult {
178178 CoordinatorMsg(Option<WorkerInput>),
179179 CollabEvent(Option<CollabEvent>),
···181181182182 loop {
183183 // Race between coordinator messages and collab events
184184- #[cfg(feature = "collab-worker")]
184184+ #[cfg(feature = "collab")]
185185 let race_result = if let Some(ref mut event_rx) = collab_event_rx {
186186 use n0_future::FutureExt;
187187 let coord_fut = async { RaceResult::CoordinatorMsg(scope.next().await) };
···191191 RaceResult::CoordinatorMsg(scope.next().await)
192192 };
193193194194- #[cfg(feature = "collab-worker")]
194194+ #[cfg(feature = "collab")]
195195 match race_result {
196196 RaceResult::CollabEvent(Some(event)) => {
197197 match event {
···281281 continue;
282282 };
283283284284- let export_start = crate::perf::now();
284284+ let export_start = weaver_common::perf::now();
285285 let snapshot_bytes = match doc.export(loro::ExportMode::Snapshot) {
286286 Ok(bytes) => bytes,
287287 Err(e) => {
···298298 continue;
299299 }
300300 };
301301- let export_ms = crate::perf::now() - export_start;
301301+ let export_ms = weaver_common::perf::now() - export_start;
302302303303- let encode_start = crate::perf::now();
303303+ let encode_start = weaver_common::perf::now();
304304 let b64_snapshot = BASE64.encode(&snapshot_bytes);
305305- let encode_ms = crate::perf::now() - encode_start;
305305+ let encode_ms = weaver_common::perf::now() - encode_start;
306306307307 let content = doc.get_text("content").to_string();
308308 let title: SmolStr = doc.get_text("title").to_string().into();
···327327 }
328328329329 // ============================================================
330330- // Collab handlers - full impl when collab-worker feature enabled
330330+ // Collab handlers - full impl when collab feature enabled
331331 // ============================================================
332332- #[cfg(feature = "collab-worker")]
332332+ #[cfg(feature = "collab")]
333333 WorkerInput::StartCollab {
334334 topic,
335335 bootstrap_peers,
···388388 "Failed to send CollabJoined to coordinator: {e}"
389389 );
390390 }
391391-392392- // NOTE: Don't broadcast Join here - wait for BroadcastJoin message
393393- // after peers have been added via AddPeers
394391395392 // Create channel for events from spawned task
396393 let (event_tx, event_rx) =
···461458 selection,
462459 ..
463460 } => {
464464- // Note: cursor updates require the collaborator to exist
465465- // (added via Join message)
466461 let exists = presence.contains(&from);
467462 tracing::debug!(%from, position, ?selection, exists, "Received Cursor message");
468463 presence.update_cursor(
···485480 }
486481 SessionEvent::PeerJoined(peer) => {
487482 tracing::info!(%peer, "PeerJoined - notifying coordinator");
488488- // Notify coordinator so it can send BroadcastJoin
489489- // Don't add to presence yet - wait for their Join message
490483 if event_tx
491484 .send(CollabEvent::PeerConnected)
492485 .is_err()
···531524 }
532525 }
533526534534- #[cfg(feature = "collab-worker")]
527527+ #[cfg(feature = "collab")]
535528 WorkerInput::BroadcastUpdate { data } => {
536529 if let Some(ref session) = collab_session {
537530 let msg = CollabMessage::LoroUpdate {
···544537 }
545538 }
546539547547- #[cfg(feature = "collab-worker")]
540540+ #[cfg(feature = "collab")]
548541 WorkerInput::BroadcastCursor {
549542 position,
550543 selection,
···572565 }
573566 }
574567575575- #[cfg(feature = "collab-worker")]
568568+ #[cfg(feature = "collab")]
576569 WorkerInput::AddPeers { peers } => {
577570 tracing::info!(count = peers.len(), "Worker: received AddPeers");
578571 if let Some(ref session) = collab_session {
579572 let peer_ids: Vec<_> = peers
580580- .iter()
581581- .filter_map(|s| {
582582- match parse_node_id(s) {
583583- Ok(id) => Some(id),
584584- Err(e) => {
585585- tracing::warn!(node_id = %s, error = %e, "Failed to parse node_id");
586586- None
587587- }
588588- }
589589- })
590590- .collect();
573573+ .iter()
574574+ .filter_map(|s| {
575575+ match parse_node_id(s) {
576576+ Ok(id) => Some(id),
577577+ Err(e) => {
578578+ tracing::warn!(node_id = %s, error = %e, "Failed to parse node_id");
579579+ None
580580+ }
581581+ }
582582+ })
583583+ .collect();
591584 tracing::info!(
592585 parsed_count = peer_ids.len(),
593586 "Worker: joining peers"
···600593 }
601594 }
602595603603- #[cfg(feature = "collab-worker")]
596596+ #[cfg(feature = "collab")]
604597 WorkerInput::BroadcastJoin { did, display_name } => {
605598 if let Some(ref session) = collab_session {
606599 let join_msg = CollabMessage::Join { did, display_name };
···610603 }
611604 }
612605613613- #[cfg(feature = "collab-worker")]
606606+ #[cfg(feature = "collab")]
614607 WorkerInput::StopCollab => {
615608 collab_session = None;
616609 collab_node = None;
···619612 tracing::error!("Failed to send CollabStopped to coordinator: {e}");
620613 }
621614 }
615615+616616+ // Non-collab stubs for when collab feature is enabled but message doesn't match
617617+ #[cfg(not(feature = "collab"))]
618618+ WorkerInput::StartCollab { .. } => {
619619+ if let Err(e) = scope
620620+ .send(WorkerOutput::Error {
621621+ message: "Collab not enabled".into(),
622622+ })
623623+ .await
624624+ {
625625+ tracing::error!("Failed to send Error to coordinator: {e}");
626626+ }
627627+ }
628628+ #[cfg(not(feature = "collab"))]
629629+ WorkerInput::BroadcastUpdate { .. } => {}
630630+ #[cfg(not(feature = "collab"))]
631631+ WorkerInput::AddPeers { .. } => {}
632632+ #[cfg(not(feature = "collab"))]
633633+ WorkerInput::BroadcastJoin { .. } => {}
634634+ #[cfg(not(feature = "collab"))]
635635+ WorkerInput::BroadcastCursor { .. } => {}
636636+ #[cfg(not(feature = "collab"))]
637637+ WorkerInput::StopCollab => {
638638+ if let Err(e) = scope.send(WorkerOutput::CollabStopped).await {
639639+ tracing::error!("Failed to send CollabStopped to coordinator: {e}");
640640+ }
641641+ }
622642 } // end match msg
623643 } // end RaceResult::CoordinatorMsg(Some(msg))
624644 } // end match race_result
625645626626- // Non-collab-worker: simple message loop
627627- #[cfg(not(feature = "collab-worker"))]
646646+ // Non-collab: simple message loop
647647+ #[cfg(not(feature = "collab"))]
628648 {
629649 let Some(msg) = scope.next().await else { break };
630650 tracing::debug!(?msg, "Worker: received message");
···679699 }
680700 continue;
681701 };
682682- let export_start = crate::perf::now();
702702+ let export_start = weaver_common::perf::now();
683703 let snapshot_bytes = match doc.export(loro::ExportMode::Snapshot) {
684704 Ok(bytes) => bytes,
685705 Err(e) => {
···696716 continue;
697717 }
698718 };
699699- let export_ms = crate::perf::now() - export_start;
700700- let encode_start = crate::perf::now();
719719+ let export_ms = weaver_common::perf::now() - export_start;
720720+ let encode_start = weaver_common::perf::now();
701721 let b64_snapshot = BASE64.encode(&snapshot_bytes);
702702- let encode_ms = crate::perf::now() - encode_start;
722722+ let encode_ms = weaver_common::perf::now() - encode_start;
703723 let content = doc.get_text("content").to_string();
704724 let title: SmolStr = doc.get_text("title").to_string().into();
705725 if let Err(e) = scope
···720740 tracing::error!("Failed to send Snapshot to coordinator: {e}");
721741 }
722742 }
723723- // Collab stubs for non-collab-worker build
743743+ // Collab stubs for non-collab build
724744 WorkerInput::StartCollab { .. } => {
725745 if let Err(e) = scope
726746 .send(WorkerOutput::Error {
···746766 }
747767748768 /// Convert PresenceTracker to serializable PresenceSnapshot.
749749- #[cfg(feature = "collab-worker")]
769769+ #[cfg(feature = "collab")]
750770 fn presence_to_snapshot(tracker: &PresenceTracker) -> PresenceSnapshot {
771771+ use jacquard::smol_str::ToSmolStr;
751772 let collaborators = tracker
752773 .collaborators()
753774 .map(|c| CollaboratorInfo {
+29
crates/weaver-common/src/agent.rs
···25892589 Ok(contributors.into_iter().collect())
25902590 }
25912591 }
25922592+25932593+ /// Fetch a blob from any PDS by DID and CID.
25942594+ ///
25952595+ /// Resolves the DID to find its PDS, then fetches the blob.
25962596+ fn fetch_blob<'a>(
25972597+ &'a self,
25982598+ did: &'a Did<'_>,
25992599+ cid: &'a jacquard::types::string::Cid<'_>,
26002600+ ) -> impl Future<Output = Result<Bytes, WeaverError>> + 'a {
26012601+ async move {
26022602+ use weaver_api::com_atproto::sync::get_blob::GetBlob;
26032603+26042604+ let pds_url = self.pds_for_did(did).await.map_err(|e| {
26052605+ AgentError::from(ClientError::from(e).with_context("Failed to resolve PDS for DID"))
26062606+ })?;
26072607+26082608+ let request = GetBlob::new().did(did.clone()).cid(cid.clone()).build();
26092609+26102610+ let response = self
26112611+ .xrpc(pds_url)
26122612+ .send(&request)
26132613+ .await
26142614+ .map_err(|e| AgentError::from(ClientError::from(e)))?;
26152615+26162616+ let output = response.into_output().map_err(|e| AgentError::xrpc(e))?;
26172617+26182618+ Ok(output.body)
26192619+ }
26202620+ }
25922621}
2593262225942623/// A version of a record from a collaborator's repository.
···11+//! Entry point for the editor web worker.
22+//!
33+//! This binary is compiled separately and loaded by the main app
44+//! to handle CPU-intensive editor operations off the main thread.
55+66+#[cfg(all(target_family = "wasm", target_os = "unknown"))]
77+fn main() {
88+ console_error_panic_hook::set_once();
99+ use tracing::Level;
1010+ use tracing::subscriber::set_global_default;
1111+ use tracing_subscriber::Registry;
1212+ use tracing_subscriber::filter::EnvFilter;
1313+ use tracing_subscriber::layer::SubscriberExt;
1414+1515+ let console_level = if cfg!(debug_assertions) {
1616+ Level::DEBUG
1717+ } else {
1818+ Level::DEBUG
1919+ };
2020+2121+ let wasm_layer = tracing_wasm::WASMLayer::new(
2222+ tracing_wasm::WASMLayerConfigBuilder::new()
2323+ .set_max_level(console_level)
2424+ .build(),
2525+ );
2626+2727+ // Filter out noisy crates
2828+ let filter = EnvFilter::new(
2929+ "debug,loro_internal=warn,jacquard_identity=info,jacquard_common=info,iroh=info",
3030+ );
3131+3232+ let reg = Registry::default().with(filter).with(wasm_layer);
3333+3434+ let _ = set_global_default(reg);
3535+3636+ use gloo_worker::Registrable;
3737+ use weaver_editor_crdt::EditorReactor;
3838+3939+ EditorReactor::registrar().register();
4040+}
4141+4242+#[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
4343+fn main() {
4444+ eprintln!("This binary is only meant to run as a WASM web worker");
4545+}
+237
crates/weaver-editor-crdt/src/buffer.rs
···11+//! Loro-backed text buffer implementing core editor traits.
22+33+use std::cell::RefCell;
44+use std::ops::Range;
55+use std::rc::Rc;
66+77+use loro::{LoroDoc, LoroText, UndoManager as LoroUndoManager, VersionVector};
88+use smol_str::{SmolStr, ToSmolStr};
99+use weaver_editor_core::{TextBuffer, UndoManager};
1010+1111+use crate::CrdtError;
1212+1313+/// Loro-backed text buffer with undo/redo support.
1414+///
1515+/// Wraps a `LoroDoc` with a text container and provides implementations
1616+/// of the `TextBuffer` and `UndoManager` traits from weaver-editor-core.
1717+#[derive(Clone)]
1818+pub struct LoroTextBuffer {
1919+ doc: LoroDoc,
2020+ content: LoroText,
2121+ undo_mgr: Rc<RefCell<LoroUndoManager>>,
2222+}
2323+2424+impl LoroTextBuffer {
2525+ /// Create a new empty buffer.
2626+ pub fn new() -> Self {
2727+ let doc = LoroDoc::new();
2828+ let content = doc.get_text("content");
2929+ let undo_mgr = Rc::new(RefCell::new(LoroUndoManager::new(&doc)));
3030+3131+ Self {
3232+ doc,
3333+ content,
3434+ undo_mgr,
3535+ }
3636+ }
3737+3838+ /// Create a buffer from an existing Loro snapshot.
3939+ pub fn from_snapshot(snapshot: &[u8]) -> Result<Self, CrdtError> {
4040+ let doc = LoroDoc::new();
4141+ doc.import(snapshot)?;
4242+ let content = doc.get_text("content");
4343+ let undo_mgr = Rc::new(RefCell::new(LoroUndoManager::new(&doc)));
4444+4545+ Ok(Self {
4646+ doc,
4747+ content,
4848+ undo_mgr,
4949+ })
5050+ }
5151+5252+ /// Get the underlying Loro document.
5353+ pub fn doc(&self) -> &LoroDoc {
5454+ &self.doc
5555+ }
5656+5757+ /// Get the text container.
5858+ pub fn content(&self) -> &LoroText {
5959+ &self.content
6060+ }
6161+6262+ /// Export full snapshot.
6363+ pub fn export_snapshot(&self) -> Vec<u8> {
6464+ self.doc
6565+ .export(loro::ExportMode::Snapshot)
6666+ .expect("snapshot export should not fail")
6767+ }
6868+6969+ /// Export updates since given version.
7070+ pub fn export_updates_since(&self, version: &VersionVector) -> Option<Vec<u8>> {
7171+ use std::borrow::Cow;
7272+7373+ let current_vv = self.doc.oplog_vv();
7474+7575+ if *version == current_vv {
7676+ return None;
7777+ }
7878+7979+ let updates = self
8080+ .doc
8181+ .export(loro::ExportMode::Updates {
8282+ from: Cow::Owned(version.clone()),
8383+ })
8484+ .ok()?;
8585+8686+ if updates.is_empty() {
8787+ return None;
8888+ }
8989+9090+ Some(updates)
9191+ }
9292+9393+ /// Import remote changes.
9494+ pub fn import(&mut self, data: &[u8]) -> Result<(), CrdtError> {
9595+ self.doc.import(data)?;
9696+ Ok(())
9797+ }
9898+9999+ /// Get current version vector.
100100+ pub fn version(&self) -> VersionVector {
101101+ self.doc.oplog_vv()
102102+ }
103103+}
104104+105105+impl Default for LoroTextBuffer {
106106+ fn default() -> Self {
107107+ Self::new()
108108+ }
109109+}
110110+111111+impl TextBuffer for LoroTextBuffer {
112112+ fn len_bytes(&self) -> usize {
113113+ self.content.to_string().len()
114114+ }
115115+116116+ fn len_chars(&self) -> usize {
117117+ self.content.len_unicode()
118118+ }
119119+120120+ fn insert(&mut self, char_offset: usize, text: &str) {
121121+ self.content.insert(char_offset, text).ok();
122122+ }
123123+124124+ fn delete(&mut self, char_range: Range<usize>) {
125125+ self.content.delete(char_range.start, char_range.len()).ok();
126126+ }
127127+128128+ fn slice(&self, char_range: Range<usize>) -> Option<SmolStr> {
129129+ let s = self.content.to_string();
130130+ let chars: Vec<char> = s.chars().collect();
131131+132132+ if char_range.end > chars.len() {
133133+ return None;
134134+ }
135135+136136+ let slice: String = chars[char_range].iter().collect();
137137+ Some(slice.to_smolstr())
138138+ }
139139+140140+ fn char_at(&self, char_offset: usize) -> Option<char> {
141141+ let s = self.content.to_string();
142142+ s.chars().nth(char_offset)
143143+ }
144144+145145+ fn to_string(&self) -> String {
146146+ self.content.to_string()
147147+ }
148148+149149+ fn char_to_byte(&self, char_offset: usize) -> usize {
150150+ let s = self.content.to_string();
151151+ s.char_indices()
152152+ .nth(char_offset)
153153+ .map(|(i, _)| i)
154154+ .unwrap_or(s.len())
155155+ }
156156+157157+ fn byte_to_char(&self, byte_offset: usize) -> usize {
158158+ let s = self.content.to_string();
159159+ s[..byte_offset.min(s.len())].chars().count()
160160+ }
161161+}
162162+163163+impl UndoManager for LoroTextBuffer {
164164+ fn can_undo(&self) -> bool {
165165+ self.undo_mgr.borrow().can_undo()
166166+ }
167167+168168+ fn can_redo(&self) -> bool {
169169+ self.undo_mgr.borrow().can_redo()
170170+ }
171171+172172+ fn undo(&mut self) -> bool {
173173+ self.undo_mgr.borrow_mut().undo().is_ok()
174174+ }
175175+176176+ fn redo(&mut self) -> bool {
177177+ self.undo_mgr.borrow_mut().redo().is_ok()
178178+ }
179179+180180+ fn clear_history(&mut self) {
181181+ // Loro's UndoManager doesn't have a clear method
182182+ // Create a new one to effectively clear history
183183+ self.undo_mgr = Rc::new(RefCell::new(LoroUndoManager::new(&self.doc)));
184184+ }
185185+}
186186+187187+#[cfg(test)]
188188+mod tests {
189189+ use super::*;
190190+191191+ #[test]
192192+ fn test_basic_operations() {
193193+ let mut buffer = LoroTextBuffer::new();
194194+195195+ buffer.insert(0, "Hello");
196196+ assert_eq!(buffer.to_string(), "Hello");
197197+198198+ buffer.insert(5, " World");
199199+ assert_eq!(buffer.to_string(), "Hello World");
200200+201201+ buffer.delete(5..6);
202202+ assert_eq!(buffer.to_string(), "HelloWorld");
203203+ }
204204+205205+ #[test]
206206+ fn test_snapshot_roundtrip() {
207207+ let mut buffer = LoroTextBuffer::new();
208208+ buffer.insert(0, "Test content");
209209+210210+ let snapshot = buffer.export_snapshot();
211211+ let restored = LoroTextBuffer::from_snapshot(&snapshot).unwrap();
212212+213213+ assert_eq!(restored.to_string(), "Test content");
214214+ }
215215+216216+ #[test]
217217+ fn test_slice() {
218218+ let mut buffer = LoroTextBuffer::new();
219219+ buffer.insert(0, "Hello World");
220220+221221+ assert_eq!(buffer.slice(0..5).as_deref(), Some("Hello"));
222222+ assert_eq!(buffer.slice(6..11).as_deref(), Some("World"));
223223+ assert_eq!(buffer.slice(0..100), None);
224224+ }
225225+226226+ #[test]
227227+ fn test_offset_conversion() {
228228+ let mut buffer = LoroTextBuffer::new();
229229+ buffer.insert(0, "hello 🌍");
230230+231231+ assert_eq!(buffer.len_chars(), 7); // h e l l o 🌍
232232+ assert_eq!(buffer.len_bytes(), 10); // 6 + 4
233233+234234+ assert_eq!(buffer.char_to_byte(6), 6); // before emoji
235235+ assert_eq!(buffer.char_to_byte(7), 10); // after emoji
236236+ }
237237+}
+159
crates/weaver-editor-crdt/src/document.rs
···11+//! CRDT document trait and sync state tracking.
22+33+use loro::VersionVector;
44+use weaver_api::com_atproto::repo::strong_ref::StrongRef;
55+66+/// Sync state for a CRDT document.
77+///
88+/// Tracks the edit root, last diff, and version at last sync.
99+#[derive(Clone, Debug, Default)]
1010+pub struct SyncState {
1111+ /// StrongRef to the sh.weaver.edit.root record.
1212+ pub edit_root: Option<StrongRef<'static>>,
1313+1414+ /// StrongRef to the most recent sh.weaver.edit.diff record.
1515+ pub last_diff: Option<StrongRef<'static>>,
1616+1717+ /// Version vector at the time of last sync.
1818+ pub last_synced_version: Option<VersionVector>,
1919+}
2020+2121+impl SyncState {
2222+ /// Create new empty sync state.
2323+ pub fn new() -> Self {
2424+ Self::default()
2525+ }
2626+2727+ /// Check if we have an edit root (i.e., have synced at least once).
2828+ pub fn has_root(&self) -> bool {
2929+ self.edit_root.is_some()
3030+ }
3131+}
3232+3333+/// Trait for CRDT documents that can be synced to AT Protocol PDS.
3434+///
3535+/// Implementors provide access to the underlying CRDT operations
3636+/// and sync state tracking.
3737+pub trait CrdtDocument {
3838+ /// Export full snapshot bytes.
3939+ fn export_snapshot(&self) -> Vec<u8>;
4040+4141+ /// Export updates since the last synced version.
4242+ /// Returns None if no changes since last sync.
4343+ fn export_updates_since_sync(&self) -> Option<Vec<u8>>;
4444+4545+ /// Import remote changes.
4646+ fn import(&mut self, data: &[u8]) -> Result<(), crate::CrdtError>;
4747+4848+ /// Get current version vector.
4949+ fn version(&self) -> VersionVector;
5050+5151+ /// Get the edit root StrongRef.
5252+ fn edit_root(&self) -> Option<StrongRef<'static>>;
5353+5454+ /// Set the edit root StrongRef.
5555+ fn set_edit_root(&mut self, root: Option<StrongRef<'static>>);
5656+5757+ /// Get the last diff StrongRef.
5858+ fn last_diff(&self) -> Option<StrongRef<'static>>;
5959+6060+ /// Set the last diff StrongRef.
6161+ fn set_last_diff(&mut self, diff: Option<StrongRef<'static>>);
6262+6363+ /// Mark current version as synced.
6464+ fn mark_synced(&mut self);
6565+6666+ /// Check if there are changes since last sync.
6767+ fn has_unsynced_changes(&self) -> bool;
6868+}
6969+7070+// Blanket implementation for LoroTextBuffer with embedded SyncState
7171+// (Concrete types can provide their own implementations)
7272+7373+/// A simple CRDT document wrapping LoroTextBuffer with sync state.
7474+pub struct SimpleCrdtDocument {
7575+ buffer: crate::LoroTextBuffer,
7676+ sync_state: SyncState,
7777+}
7878+7979+impl SimpleCrdtDocument {
8080+ /// Create a new empty document.
8181+ pub fn new() -> Self {
8282+ Self {
8383+ buffer: crate::LoroTextBuffer::new(),
8484+ sync_state: SyncState::new(),
8585+ }
8686+ }
8787+8888+ /// Create from snapshot.
8989+ pub fn from_snapshot(snapshot: &[u8]) -> Result<Self, crate::CrdtError> {
9090+ Ok(Self {
9191+ buffer: crate::LoroTextBuffer::from_snapshot(snapshot)?,
9292+ sync_state: SyncState::new(),
9393+ })
9494+ }
9595+9696+ /// Get the underlying buffer.
9797+ pub fn buffer(&self) -> &crate::LoroTextBuffer {
9898+ &self.buffer
9999+ }
100100+101101+ /// Get mutable access to the buffer.
102102+ pub fn buffer_mut(&mut self) -> &mut crate::LoroTextBuffer {
103103+ &mut self.buffer
104104+ }
105105+}
106106+107107+impl Default for SimpleCrdtDocument {
108108+ fn default() -> Self {
109109+ Self::new()
110110+ }
111111+}
112112+113113+impl CrdtDocument for SimpleCrdtDocument {
114114+ fn export_snapshot(&self) -> Vec<u8> {
115115+ self.buffer.export_snapshot()
116116+ }
117117+118118+ fn export_updates_since_sync(&self) -> Option<Vec<u8>> {
119119+ self.sync_state
120120+ .last_synced_version
121121+ .as_ref()
122122+ .and_then(|v| self.buffer.export_updates_since(v))
123123+ }
124124+125125+ fn import(&mut self, data: &[u8]) -> Result<(), crate::CrdtError> {
126126+ self.buffer.import(data)
127127+ }
128128+129129+ fn version(&self) -> VersionVector {
130130+ self.buffer.version()
131131+ }
132132+133133+ fn edit_root(&self) -> Option<StrongRef<'static>> {
134134+ self.sync_state.edit_root.clone()
135135+ }
136136+137137+ fn set_edit_root(&mut self, root: Option<StrongRef<'static>>) {
138138+ self.sync_state.edit_root = root;
139139+ }
140140+141141+ fn last_diff(&self) -> Option<StrongRef<'static>> {
142142+ self.sync_state.last_diff.clone()
143143+ }
144144+145145+ fn set_last_diff(&mut self, diff: Option<StrongRef<'static>>) {
146146+ self.sync_state.last_diff = diff;
147147+ }
148148+149149+ fn mark_synced(&mut self) {
150150+ self.sync_state.last_synced_version = Some(self.buffer.version());
151151+ }
152152+153153+ fn has_unsynced_changes(&self) -> bool {
154154+ match &self.sync_state.last_synced_version {
155155+ None => true, // Never synced
156156+ Some(last) => self.buffer.version() != *last,
157157+ }
158158+ }
159159+}
···11+//! CRDT-backed editor with AT Protocol sync.
22+//!
33+//! This crate provides:
44+//! - `LoroTextBuffer`: Loro-backed text buffer implementing `TextBuffer` + `UndoManager`
55+//! - `CrdtDocument`: Trait for documents that can sync to AT Protocol PDS
66+//! - Generic sync logic for edit records (root/diff/draft)
77+//! - Worker implementation for off-main-thread CRDT operations
88+99+mod buffer;
1010+mod document;
1111+mod error;
1212+mod sync;
1313+1414+pub mod worker;
1515+1616+pub use buffer::LoroTextBuffer;
1717+pub use document::{CrdtDocument, SimpleCrdtDocument, SyncState};
1818+pub use error::CrdtError;
1919+pub use sync::{
2020+ CreateRootResult, PdsEditState, RemoteDraft, SyncResult,
2121+ build_draft_uri, create_diff, create_edit_root,
2222+ find_all_edit_roots, find_diffs_for_root, find_edit_root_for_draft,
2323+ list_drafts, load_all_edit_states, load_edit_state_from_draft,
2424+ load_edit_state_from_entry, sync_to_pds,
2525+};
2626+2727+// Re-export worker types
2828+pub use worker::{WorkerInput, WorkerOutput};
2929+#[cfg(all(target_family = "wasm", target_os = "unknown"))]
3030+pub use worker::EditorReactor;
3131+3232+// Re-export Loro types that consumers need
3333+pub use loro::{ExportMode, LoroDoc, LoroText, VersionVector};
+963
crates/weaver-editor-crdt/src/sync.rs
···11+//! PDS synchronization for CRDT documents.
22+//!
33+//! Generic sync logic for AT Protocol edit records (root/diff/draft).
44+//! Works with any client implementing the required traits.
55+66+use std::collections::{BTreeMap, HashMap};
77+88+use jacquard::bytes::Bytes;
99+use jacquard::cowstr::ToCowStr;
1010+use jacquard::prelude::*;
1111+use jacquard::smol_str::format_smolstr;
1212+use jacquard::types::blob::MimeType;
1313+use jacquard::types::ident::AtIdentifier;
1414+use jacquard::types::recordkey::RecordKey;
1515+use jacquard::types::string::{AtUri, Cid, Did, Nsid};
1616+use jacquard::types::tid::Ticker;
1717+use jacquard::types::uri::Uri;
1818+use jacquard::url::Url;
1919+use jacquard::{CowStr, IntoStatic, to_data};
2020+use loro::{ExportMode, LoroDoc};
2121+use weaver_api::com_atproto::repo::create_record::CreateRecord;
2222+use weaver_api::com_atproto::repo::strong_ref::StrongRef;
2323+use weaver_api::sh_weaver::edit::diff::Diff;
2424+use weaver_api::sh_weaver::edit::draft::Draft;
2525+use weaver_api::sh_weaver::edit::root::Root;
2626+use weaver_api::sh_weaver::edit::{DocRef, DocRefValue, DraftRef, EntryRef};
2727+use weaver_common::agent::WeaverExt;
2828+use weaver_common::constellation::{GetBacklinksQuery, RecordId};
2929+3030+use crate::CrdtError;
3131+use crate::document::CrdtDocument;
3232+3333+const ROOT_NSID: &str = "sh.weaver.edit.root";
3434+const DIFF_NSID: &str = "sh.weaver.edit.diff";
3535+const DRAFT_NSID: &str = "sh.weaver.edit.draft";
3636+const CONSTELLATION_URL: &str = "https://constellation.microcosm.blue";
3737+3838+/// Result of a sync operation.
3939+#[derive(Clone, Debug)]
4040+pub enum SyncResult {
4141+ /// Created a new root record (first sync).
4242+ CreatedRoot {
4343+ uri: AtUri<'static>,
4444+ cid: Cid<'static>,
4545+ },
4646+ /// Created a new diff record.
4747+ CreatedDiff {
4848+ uri: AtUri<'static>,
4949+ cid: Cid<'static>,
5050+ },
5151+ /// No changes to sync.
5252+ NoChanges,
5353+}
5454+5555+/// Result of creating an edit root.
5656+pub struct CreateRootResult {
5757+ /// The root record URI.
5858+ pub root_uri: AtUri<'static>,
5959+ /// The root record CID.
6060+ pub root_cid: Cid<'static>,
6161+ /// Draft stub StrongRef if this was a new draft.
6262+ pub draft_ref: Option<StrongRef<'static>>,
6363+}
6464+6565+/// Build a DocRef for either a published entry or an unpublished draft.
6666+fn build_doc_ref(
6767+ did: &Did<'_>,
6868+ draft_key: &str,
6969+ entry_uri: Option<&AtUri<'_>>,
7070+ entry_cid: Option<&Cid<'_>>,
7171+) -> DocRef<'static> {
7272+ match (entry_uri, entry_cid) {
7373+ (Some(uri), Some(cid)) => DocRef {
7474+ value: DocRefValue::EntryRef(Box::new(EntryRef {
7575+ entry: StrongRef::new()
7676+ .uri(uri.clone().into_static())
7777+ .cid(cid.clone().into_static())
7878+ .build(),
7979+ extra_data: None,
8080+ })),
8181+ extra_data: None,
8282+ },
8383+ _ => {
8484+ // Transform localStorage key to synthetic AT-URI
8585+ let rkey = extract_draft_rkey(draft_key);
8686+ let canonical_uri = format_smolstr!("at://{}/{}/{}", did, DRAFT_NSID, rkey);
8787+8888+ DocRef {
8989+ value: DocRefValue::DraftRef(Box::new(DraftRef {
9090+ draft_key: canonical_uri.into(),
9191+ extra_data: None,
9292+ })),
9393+ extra_data: None,
9494+ }
9595+ }
9696+ }
9797+}
9898+9999+/// Extract the rkey (TID) from a draft key.
100100+fn extract_draft_rkey(draft_key: &str) -> String {
101101+ if let Some(tid) = draft_key.strip_prefix("new:") {
102102+ tid.to_string()
103103+ } else if draft_key.starts_with("at://") {
104104+ draft_key.split('/').last().unwrap_or(draft_key).to_string()
105105+ } else {
106106+ draft_key.to_string()
107107+ }
108108+}
109109+110110+/// Get current DID from session.
111111+async fn get_current_did<C>(client: &C) -> Result<Did<'static>, CrdtError>
112112+where
113113+ C: AgentSession,
114114+{
115115+ client
116116+ .session_info()
117117+ .await
118118+ .map(|(did, _)| did)
119119+ .ok_or(CrdtError::NotAuthenticated)
120120+}
121121+122122+/// Create the draft stub record on PDS.
123123+async fn create_draft_stub<C>(
124124+ client: &C,
125125+ did: &Did<'_>,
126126+ rkey: &str,
127127+) -> Result<(AtUri<'static>, Cid<'static>), CrdtError>
128128+where
129129+ C: XrpcClient + AgentSession,
130130+{
131131+ let draft = Draft::new()
132132+ .created_at(jacquard::types::datetime::Datetime::now())
133133+ .build();
134134+135135+ let draft_data =
136136+ to_data(&draft).map_err(|e| CrdtError::Serialization(format!("draft: {}", e)))?;
137137+138138+ let record_key =
139139+ RecordKey::any(rkey).map_err(|e| CrdtError::InvalidUri(format!("rkey: {}", e)))?;
140140+141141+ let collection =
142142+ Nsid::new(DRAFT_NSID).map_err(|e| CrdtError::InvalidUri(format!("nsid: {}", e)))?;
143143+144144+ let request = CreateRecord::new()
145145+ .repo(AtIdentifier::Did(did.clone().into_static()))
146146+ .collection(collection)
147147+ .rkey(record_key)
148148+ .record(draft_data)
149149+ .build();
150150+151151+ let response = client
152152+ .send(request)
153153+ .await
154154+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
155155+156156+ let output = response
157157+ .into_output()
158158+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
159159+160160+ Ok((output.uri.into_static(), output.cid.into_static()))
161161+}
162162+163163+/// Create the edit root record for a document.
164164+pub async fn create_edit_root<C, D>(
165165+ client: &C,
166166+ doc: &D,
167167+ draft_key: &str,
168168+ entry_uri: Option<&AtUri<'_>>,
169169+ entry_cid: Option<&Cid<'_>>,
170170+) -> Result<CreateRootResult, CrdtError>
171171+where
172172+ C: XrpcClient + IdentityResolver + AgentSession,
173173+ D: CrdtDocument,
174174+{
175175+ let did = get_current_did(client).await?;
176176+177177+ // For drafts, create the stub record first
178178+ let draft_ref: Option<StrongRef<'static>> = if entry_uri.is_none() {
179179+ let rkey = extract_draft_rkey(draft_key);
180180+ match create_draft_stub(client, &did, &rkey).await {
181181+ Ok((uri, cid)) => Some(StrongRef::new().uri(uri).cid(cid).build()),
182182+ Err(e) => {
183183+ let err_str = e.to_string();
184184+ if err_str.contains("RecordAlreadyExists") || err_str.contains("already exists") {
185185+ // Draft exists, try to fetch it
186186+ let draft_uri_str = format!("at://{}/{}/{}", did, DRAFT_NSID, rkey);
187187+ if let Ok(draft_uri) = AtUri::new(&draft_uri_str) {
188188+ if let Ok(response) = client.get_record::<Draft>(&draft_uri).await {
189189+ if let Ok(output) = response.into_output() {
190190+ output.cid.map(|cid| {
191191+ StrongRef::new()
192192+ .uri(draft_uri.into_static())
193193+ .cid(cid.into_static())
194194+ .build()
195195+ })
196196+ } else {
197197+ None
198198+ }
199199+ } else {
200200+ None
201201+ }
202202+ } else {
203203+ None
204204+ }
205205+ } else {
206206+ tracing::warn!("Failed to create draft stub: {}", e);
207207+ None
208208+ }
209209+ }
210210+ }
211211+ } else {
212212+ None
213213+ };
214214+215215+ // Export full snapshot
216216+ let snapshot = doc.export_snapshot();
217217+218218+ // Upload snapshot blob
219219+ let mime_type = MimeType::new_static("application/octet-stream");
220220+ let blob_ref = client
221221+ .upload_blob(snapshot, mime_type)
222222+ .await
223223+ .map_err(|e| CrdtError::Xrpc(format!("upload blob: {}", e)))?;
224224+225225+ // Build DocRef
226226+ let doc_ref = build_doc_ref(&did, draft_key, entry_uri, entry_cid);
227227+228228+ // Build root record
229229+ let root = Root::new().doc(doc_ref).snapshot(blob_ref).build();
230230+231231+ let root_data = to_data(&root).map_err(|e| CrdtError::Serialization(format!("root: {}", e)))?;
232232+233233+ // Generate TID for the root rkey
234234+ let root_tid = Ticker::new().next(None);
235235+ let rkey = RecordKey::any(root_tid.as_str())
236236+ .map_err(|e| CrdtError::InvalidUri(format!("rkey: {}", e)))?;
237237+238238+ let collection =
239239+ Nsid::new(ROOT_NSID).map_err(|e| CrdtError::InvalidUri(format!("nsid: {}", e)))?;
240240+241241+ let request = CreateRecord::new()
242242+ .repo(AtIdentifier::Did(did))
243243+ .collection(collection)
244244+ .rkey(rkey)
245245+ .record(root_data)
246246+ .build();
247247+248248+ let response = client
249249+ .send(request)
250250+ .await
251251+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
252252+253253+ let output = response
254254+ .into_output()
255255+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
256256+257257+ Ok(CreateRootResult {
258258+ root_uri: output.uri.into_static(),
259259+ root_cid: output.cid.into_static(),
260260+ draft_ref,
261261+ })
262262+}
263263+264264+/// Create a diff record with updates since the last sync.
265265+pub async fn create_diff<C, D>(
266266+ client: &C,
267267+ doc: &D,
268268+ root_uri: &AtUri<'_>,
269269+ root_cid: &Cid<'_>,
270270+ prev_diff: Option<(&AtUri<'_>, &Cid<'_>)>,
271271+ draft_key: &str,
272272+ entry_uri: Option<&AtUri<'_>>,
273273+ entry_cid: Option<&Cid<'_>>,
274274+) -> Result<Option<(AtUri<'static>, Cid<'static>)>, CrdtError>
275275+where
276276+ C: XrpcClient + IdentityResolver + AgentSession,
277277+ D: CrdtDocument,
278278+{
279279+ // Export updates since last sync
280280+ let updates = match doc.export_updates_since_sync() {
281281+ Some(u) => u,
282282+ None => return Ok(None),
283283+ };
284284+285285+ let did = get_current_did(client).await?;
286286+287287+ // Threshold for inline vs blob storage (8KB max for inline per lexicon)
288288+ const INLINE_THRESHOLD: usize = 8192;
289289+290290+ let (blob_ref, inline_diff): (Option<jacquard::types::blob::BlobRef<'static>>, _) =
291291+ if updates.len() <= INLINE_THRESHOLD {
292292+ (None, Some(jacquard::bytes::Bytes::from(updates)))
293293+ } else {
294294+ let mime_type = MimeType::new_static("application/octet-stream");
295295+ let blob = client
296296+ .upload_blob(updates, mime_type)
297297+ .await
298298+ .map_err(|e| CrdtError::Xrpc(format!("upload diff: {}", e)))?;
299299+ (Some(blob.into()), None)
300300+ };
301301+302302+ // Build DocRef
303303+ let doc_ref = build_doc_ref(&did, draft_key, entry_uri, entry_cid);
304304+305305+ // Build root reference
306306+ let root_ref = StrongRef::new()
307307+ .uri(root_uri.clone().into_static())
308308+ .cid(root_cid.clone().into_static())
309309+ .build();
310310+311311+ // Build prev reference
312312+ let prev_ref = prev_diff.map(|(uri, cid)| {
313313+ StrongRef::new()
314314+ .uri(uri.clone().into_static())
315315+ .cid(cid.clone().into_static())
316316+ .build()
317317+ });
318318+319319+ // Build diff record
320320+ let diff = Diff::new()
321321+ .doc(doc_ref)
322322+ .root(root_ref)
323323+ .maybe_snapshot(blob_ref)
324324+ .maybe_inline_diff(inline_diff)
325325+ .maybe_prev(prev_ref)
326326+ .build();
327327+328328+ let diff_data = to_data(&diff).map_err(|e| CrdtError::Serialization(format!("diff: {}", e)))?;
329329+330330+ // Generate TID for the diff rkey
331331+ let diff_tid = Ticker::new().next(None);
332332+ let rkey = RecordKey::any(diff_tid.as_str())
333333+ .map_err(|e| CrdtError::InvalidUri(format!("rkey: {}", e)))?;
334334+335335+ let collection =
336336+ Nsid::new(DIFF_NSID).map_err(|e| CrdtError::InvalidUri(format!("nsid: {}", e)))?;
337337+338338+ let request = CreateRecord::new()
339339+ .repo(AtIdentifier::Did(did))
340340+ .collection(collection)
341341+ .rkey(rkey)
342342+ .record(diff_data)
343343+ .build();
344344+345345+ let response = client
346346+ .send(request)
347347+ .await
348348+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
349349+350350+ let output = response
351351+ .into_output()
352352+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
353353+354354+ Ok(Some((output.uri.into_static(), output.cid.into_static())))
355355+}
356356+357357+/// Sync the document to the PDS.
358358+pub async fn sync_to_pds<C, D>(
359359+ client: &C,
360360+ doc: &mut D,
361361+ draft_key: &str,
362362+ entry_uri: Option<&AtUri<'_>>,
363363+ entry_cid: Option<&Cid<'_>>,
364364+) -> Result<SyncResult, CrdtError>
365365+where
366366+ C: XrpcClient + IdentityResolver + AgentSession,
367367+ D: CrdtDocument,
368368+{
369369+ if !doc.has_unsynced_changes() {
370370+ return Ok(SyncResult::NoChanges);
371371+ }
372372+373373+ if doc.edit_root().is_none() {
374374+ // First sync - create root
375375+ let result = create_edit_root(client, doc, draft_key, entry_uri, entry_cid).await?;
376376+377377+ let root_ref = StrongRef::new()
378378+ .uri(result.root_uri.clone())
379379+ .cid(result.root_cid.clone())
380380+ .build();
381381+382382+ doc.set_edit_root(Some(root_ref));
383383+ doc.set_last_diff(None);
384384+ doc.mark_synced();
385385+386386+ Ok(SyncResult::CreatedRoot {
387387+ uri: result.root_uri,
388388+ cid: result.root_cid,
389389+ })
390390+ } else {
391391+ // Subsequent sync - create diff
392392+ let root = doc.edit_root().unwrap();
393393+ let prev = doc.last_diff();
394394+395395+ let prev_refs = prev.as_ref().map(|p| (&p.uri, &p.cid));
396396+397397+ let result = create_diff(
398398+ client, doc, &root.uri, &root.cid, prev_refs, draft_key, entry_uri, entry_cid,
399399+ )
400400+ .await?;
401401+402402+ match result {
403403+ Some((uri, cid)) => {
404404+ let diff_ref = StrongRef::new().uri(uri.clone()).cid(cid.clone()).build();
405405+ doc.set_last_diff(Some(diff_ref));
406406+ doc.mark_synced();
407407+408408+ Ok(SyncResult::CreatedDiff { uri, cid })
409409+ }
410410+ None => Ok(SyncResult::NoChanges),
411411+ }
412412+ }
413413+}
414414+415415+/// Find all edit roots for an entry using weaver-index.
416416+#[cfg(feature = "use-index")]
417417+pub async fn find_all_edit_roots<C>(
418418+ client: &C,
419419+ entry_uri: &AtUri<'_>,
420420+ _collaborator_dids: Vec<Did<'static>>,
421421+) -> Result<Vec<RecordId<'static>>, CrdtError>
422422+where
423423+ C: WeaverExt,
424424+{
425425+ use jacquard::types::ident::AtIdentifier;
426426+ use jacquard::types::nsid::Nsid;
427427+ use weaver_api::sh_weaver::edit::get_edit_history::GetEditHistory;
428428+429429+ let response = client
430430+ .send(GetEditHistory::new().resource(entry_uri.clone()).build())
431431+ .await
432432+ .map_err(|e| CrdtError::Xrpc(format!("get edit history: {}", e)))?;
433433+434434+ let output = response
435435+ .into_output()
436436+ .map_err(|e| CrdtError::Xrpc(format!("parse edit history: {}", e)))?;
437437+438438+ let roots: Vec<RecordId<'static>> = output
439439+ .roots
440440+ .into_iter()
441441+ .filter_map(|entry| {
442442+ let uri = AtUri::new(entry.uri.as_ref()).ok()?;
443443+ let did = match uri.authority() {
444444+ AtIdentifier::Did(d) => d.clone().into_static(),
445445+ _ => return None,
446446+ };
447447+ let rkey = uri.rkey()?.clone().into_static();
448448+ Some(RecordId {
449449+ did,
450450+ collection: Nsid::raw(ROOT_NSID).into_static(),
451451+ rkey,
452452+ })
453453+ })
454454+ .collect();
455455+456456+ tracing::debug!("find_all_edit_roots (index): found {} roots", roots.len());
457457+458458+ Ok(roots)
459459+}
460460+461461+/// Find all edit roots for an entry using Constellation backlinks.
462462+#[cfg(not(feature = "use-index"))]
463463+pub async fn find_all_edit_roots<C>(
464464+ client: &C,
465465+ entry_uri: &AtUri<'_>,
466466+ collaborator_dids: Vec<Did<'static>>,
467467+) -> Result<Vec<RecordId<'static>>, CrdtError>
468468+where
469469+ C: XrpcClient,
470470+{
471471+ let constellation_url =
472472+ Url::parse(CONSTELLATION_URL).map_err(|e| CrdtError::InvalidUri(e.to_string()))?;
473473+474474+ let query = GetBacklinksQuery {
475475+ subject: Uri::At(entry_uri.clone().into_static()),
476476+ source: format_smolstr!("{}:doc.value.entry.uri", ROOT_NSID).into(),
477477+ cursor: None,
478478+ did: collaborator_dids,
479479+ limit: 100,
480480+ };
481481+482482+ let response = client
483483+ .xrpc(constellation_url)
484484+ .send(&query)
485485+ .await
486486+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
487487+488488+ let output = response
489489+ .into_output()
490490+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
491491+492492+ Ok(output
493493+ .records
494494+ .into_iter()
495495+ .map(|r| r.into_static())
496496+ .collect())
497497+}
498498+499499+/// Find all diffs for a root record using Constellation backlinks.
500500+pub async fn find_diffs_for_root<C>(
501501+ client: &C,
502502+ root_uri: &AtUri<'_>,
503503+) -> Result<Vec<RecordId<'static>>, CrdtError>
504504+where
505505+ C: XrpcClient,
506506+{
507507+ let constellation_url =
508508+ Url::parse(CONSTELLATION_URL).map_err(|e| CrdtError::InvalidUri(e.to_string()))?;
509509+510510+ let mut all_diffs = Vec::new();
511511+ let mut cursor: Option<String> = None;
512512+513513+ loop {
514514+ let query = GetBacklinksQuery {
515515+ subject: Uri::At(root_uri.clone().into_static()),
516516+ source: format_smolstr!("{}:root.uri", DIFF_NSID).into(),
517517+ cursor: cursor.map(Into::into),
518518+ did: vec![],
519519+ limit: 100,
520520+ };
521521+522522+ let response = client
523523+ .xrpc(constellation_url.clone())
524524+ .send(&query)
525525+ .await
526526+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
527527+528528+ let output = response
529529+ .into_output()
530530+ .map_err(|e| CrdtError::Xrpc(e.to_string()))?;
531531+532532+ all_diffs.extend(output.records.into_iter().map(|r| r.into_static()));
533533+534534+ match output.cursor {
535535+ Some(c) => cursor = Some(c.to_string()),
536536+ None => break,
537537+ }
538538+ }
539539+540540+ Ok(all_diffs)
541541+}
542542+543543+// ============================================================================
544544+// Loading functions
545545+// ============================================================================
546546+547547+/// Result of loading edit state from PDS.
548548+#[derive(Clone, Debug)]
549549+pub struct PdsEditState {
550550+ /// The root record reference.
551551+ pub root_ref: StrongRef<'static>,
552552+ /// The latest diff reference (if any diffs exist).
553553+ pub last_diff_ref: Option<StrongRef<'static>>,
554554+ /// The Loro snapshot bytes from the root.
555555+ pub root_snapshot: Bytes,
556556+ /// All diff update bytes in order (oldest first, by TID).
557557+ pub diff_updates: Vec<Bytes>,
558558+ /// Last seen diff URI per collaborator root (for incremental sync).
559559+ pub last_seen_diffs: HashMap<AtUri<'static>, AtUri<'static>>,
560560+ /// The DocRef from the root record.
561561+ pub doc_ref: DocRef<'static>,
562562+}
563563+564564+/// Find edit root for a draft using Constellation backlinks.
565565+pub async fn find_edit_root_for_draft<C>(
566566+ client: &C,
567567+ draft_uri: &AtUri<'_>,
568568+) -> Result<Option<RecordId<'static>>, CrdtError>
569569+where
570570+ C: XrpcClient,
571571+{
572572+ let constellation_url =
573573+ Url::parse(CONSTELLATION_URL).map_err(|e| CrdtError::InvalidUri(e.to_string()))?;
574574+575575+ let query = GetBacklinksQuery {
576576+ subject: Uri::At(draft_uri.clone().into_static()),
577577+ source: format_smolstr!("{}:doc.value.draft_key", ROOT_NSID).into(),
578578+ cursor: None,
579579+ did: vec![],
580580+ limit: 1,
581581+ };
582582+583583+ let response = client
584584+ .xrpc(constellation_url)
585585+ .send(&query)
586586+ .await
587587+ .map_err(|e| CrdtError::Xrpc(format!("constellation query: {}", e)))?;
588588+589589+ let output = response
590590+ .into_output()
591591+ .map_err(|e| CrdtError::Xrpc(format!("parse constellation: {}", e)))?;
592592+593593+ Ok(output.records.into_iter().next().map(|r| r.into_static()))
594594+}
595595+596596+/// Build a canonical draft URI from draft key and DID.
597597+pub fn build_draft_uri(did: &Did<'_>, draft_key: &str) -> AtUri<'static> {
598598+ let rkey = extract_draft_rkey(draft_key);
599599+ let uri_str = format_smolstr!("at://{}/{}/{}", did, DRAFT_NSID, rkey);
600600+ AtUri::new(&uri_str).unwrap().into_static()
601601+}
602602+603603+/// Load edit state from a root record ID.
604604+async fn load_edit_state_from_root_id<C>(
605605+ client: &C,
606606+ root_id: RecordId<'static>,
607607+ after_rkey: Option<&str>,
608608+) -> Result<Option<PdsEditState>, CrdtError>
609609+where
610610+ C: WeaverExt,
611611+{
612612+ let root_uri = AtUri::new(&format_smolstr!(
613613+ "at://{}/{}/{}",
614614+ root_id.did,
615615+ ROOT_NSID,
616616+ root_id.rkey.as_ref()
617617+ ))
618618+ .map_err(|e| CrdtError::InvalidUri(format!("root URI: {}", e)))?
619619+ .into_static();
620620+621621+ let root_response = client
622622+ .get_record::<Root>(&root_uri)
623623+ .await
624624+ .map_err(|e| CrdtError::Xrpc(format!("fetch root: {}", e)))?;
625625+626626+ let root_output = root_response
627627+ .into_output()
628628+ .map_err(|e| CrdtError::Xrpc(format!("parse root: {}", e)))?;
629629+630630+ let root_cid = root_output
631631+ .cid
632632+ .ok_or_else(|| CrdtError::Xrpc("root missing CID".into()))?;
633633+634634+ let root_ref = StrongRef::new()
635635+ .uri(root_uri.clone())
636636+ .cid(root_cid.into_static())
637637+ .build();
638638+639639+ let doc_ref = root_output.value.doc.into_static();
640640+641641+ let root_snapshot = client
642642+ .fetch_blob(&root_id.did, root_output.value.snapshot.blob().cid())
643643+ .await
644644+ .map_err(|e| CrdtError::Xrpc(format!("fetch snapshot blob: {}", e)))?;
645645+646646+ let diff_ids = find_diffs_for_root(client, &root_uri).await?;
647647+648648+ if diff_ids.is_empty() {
649649+ return Ok(Some(PdsEditState {
650650+ root_ref,
651651+ last_diff_ref: None,
652652+ root_snapshot,
653653+ diff_updates: vec![],
654654+ last_seen_diffs: HashMap::new(),
655655+ doc_ref,
656656+ }));
657657+ }
658658+659659+ let mut diffs_by_rkey: BTreeMap<
660660+ CowStr<'static>,
661661+ (Diff<'static>, Cid<'static>, AtUri<'static>),
662662+ > = BTreeMap::new();
663663+664664+ for diff_id in &diff_ids {
665665+ let rkey_str: &str = diff_id.rkey.as_ref();
666666+667667+ if let Some(after) = after_rkey {
668668+ if rkey_str <= after {
669669+ continue;
670670+ }
671671+ }
672672+673673+ let diff_uri = AtUri::new(&format_smolstr!(
674674+ "at://{}/{}/{}",
675675+ diff_id.did,
676676+ DIFF_NSID,
677677+ rkey_str
678678+ ))
679679+ .map_err(|e| CrdtError::InvalidUri(format!("diff URI: {}", e)))?
680680+ .into_static();
681681+682682+ let diff_response = client
683683+ .get_record::<Diff>(&diff_uri)
684684+ .await
685685+ .map_err(|e| CrdtError::Xrpc(format!("fetch diff: {}", e)))?;
686686+687687+ let diff_output = diff_response
688688+ .into_output()
689689+ .map_err(|e| CrdtError::Xrpc(format!("parse diff: {}", e)))?;
690690+691691+ let diff_cid = diff_output
692692+ .cid
693693+ .ok_or_else(|| CrdtError::Xrpc("diff missing CID".into()))?;
694694+695695+ diffs_by_rkey.insert(
696696+ rkey_str.to_cowstr().into_static(),
697697+ (
698698+ diff_output.value.into_static(),
699699+ diff_cid.into_static(),
700700+ diff_uri,
701701+ ),
702702+ );
703703+ }
704704+705705+ let mut diff_updates = Vec::new();
706706+ let mut last_diff_ref = None;
707707+708708+ for (_rkey, (diff, cid, uri)) in &diffs_by_rkey {
709709+ let diff_bytes = if let Some(ref inline) = diff.inline_diff {
710710+ inline.clone()
711711+ } else if let Some(ref snapshot) = diff.snapshot {
712712+ client
713713+ .fetch_blob(&root_id.did, snapshot.blob().cid())
714714+ .await
715715+ .map_err(|e| CrdtError::Xrpc(format!("fetch diff blob: {}", e)))?
716716+ } else {
717717+ tracing::warn!("Diff has neither inline_diff nor snapshot, skipping");
718718+ continue;
719719+ };
720720+721721+ diff_updates.push(diff_bytes);
722722+ last_diff_ref = Some(StrongRef::new().uri(uri.clone()).cid(cid.clone()).build());
723723+ }
724724+725725+ Ok(Some(PdsEditState {
726726+ root_ref,
727727+ last_diff_ref,
728728+ root_snapshot,
729729+ diff_updates,
730730+ last_seen_diffs: HashMap::new(),
731731+ doc_ref,
732732+ }))
733733+}
734734+735735+/// Load edit state from PDS for an entry (single root).
736736+pub async fn load_edit_state_from_entry<C>(
737737+ client: &C,
738738+ entry_uri: &AtUri<'_>,
739739+ collaborator_dids: Vec<Did<'static>>,
740740+) -> Result<Option<PdsEditState>, CrdtError>
741741+where
742742+ C: WeaverExt,
743743+{
744744+ let root_id = match find_all_edit_roots(client, entry_uri, collaborator_dids)
745745+ .await?
746746+ .into_iter()
747747+ .next()
748748+ {
749749+ Some(id) => id,
750750+ None => return Ok(None),
751751+ };
752752+753753+ load_edit_state_from_root_id(client, root_id, None).await
754754+}
755755+756756+/// Load edit state from PDS for a draft.
757757+pub async fn load_edit_state_from_draft<C>(
758758+ client: &C,
759759+ draft_uri: &AtUri<'_>,
760760+) -> Result<Option<PdsEditState>, CrdtError>
761761+where
762762+ C: WeaverExt,
763763+{
764764+ let root_id = match find_edit_root_for_draft(client, draft_uri).await? {
765765+ Some(id) => id,
766766+ None => return Ok(None),
767767+ };
768768+769769+ load_edit_state_from_root_id(client, root_id, None).await
770770+}
771771+772772+/// Load and merge edit states from ALL collaborator repos.
773773+pub async fn load_all_edit_states<C>(
774774+ client: &C,
775775+ entry_uri: &AtUri<'_>,
776776+ collaborator_dids: Vec<Did<'static>>,
777777+ current_did: Option<&Did<'_>>,
778778+ last_seen_diffs: &HashMap<AtUri<'static>, AtUri<'static>>,
779779+) -> Result<Option<PdsEditState>, CrdtError>
780780+where
781781+ C: WeaverExt,
782782+{
783783+ let all_roots = find_all_edit_roots(client, entry_uri, collaborator_dids).await?;
784784+785785+ if all_roots.is_empty() {
786786+ return Ok(None);
787787+ }
788788+789789+ let merged_doc = LoroDoc::new();
790790+ let mut our_root_ref: Option<StrongRef<'static>> = None;
791791+ let mut our_last_diff_ref: Option<StrongRef<'static>> = None;
792792+ let mut merged_doc_ref: Option<DocRef<'static>> = None;
793793+ let mut updated_last_seen = last_seen_diffs.clone();
794794+795795+ for root_id in all_roots {
796796+ let root_did = root_id.did.clone();
797797+798798+ let root_uri = AtUri::new(&format_smolstr!(
799799+ "at://{}/{}/{}",
800800+ root_id.did,
801801+ ROOT_NSID,
802802+ root_id.rkey.as_ref()
803803+ ))
804804+ .ok()
805805+ .map(|u| u.into_static());
806806+807807+ let after_rkey = root_uri.as_ref().and_then(|uri| {
808808+ last_seen_diffs
809809+ .get(uri)
810810+ .and_then(|diff_uri| diff_uri.rkey().map(|rk| rk.0.to_string()))
811811+ });
812812+813813+ if let Some(pds_state) =
814814+ load_edit_state_from_root_id(client, root_id, after_rkey.as_deref()).await?
815815+ {
816816+ if let Err(e) = merged_doc.import(&pds_state.root_snapshot) {
817817+ tracing::warn!("Failed to import root snapshot from {}: {:?}", root_did, e);
818818+ continue;
819819+ }
820820+821821+ for diff in &pds_state.diff_updates {
822822+ if let Err(e) = merged_doc.import(diff) {
823823+ tracing::warn!("Failed to import diff from {}: {:?}", root_did, e);
824824+ }
825825+ }
826826+827827+ if let (Some(uri), Some(last_diff)) = (&root_uri, &pds_state.last_diff_ref) {
828828+ updated_last_seen.insert(uri.clone(), last_diff.uri.clone().into_static());
829829+ }
830830+831831+ if merged_doc_ref.is_none() {
832832+ merged_doc_ref = Some(pds_state.doc_ref.clone());
833833+ }
834834+835835+ let is_our_root = current_did.is_some_and(|did| root_did == *did);
836836+837837+ if is_our_root {
838838+ our_root_ref = Some(pds_state.root_ref);
839839+ our_last_diff_ref = pds_state.last_diff_ref;
840840+ } else if our_root_ref.is_none() {
841841+ our_root_ref = Some(pds_state.root_ref);
842842+ our_last_diff_ref = pds_state.last_diff_ref;
843843+ }
844844+ }
845845+ }
846846+847847+ let merged_snapshot = merged_doc
848848+ .export(ExportMode::Snapshot)
849849+ .map_err(|e| CrdtError::Loro(format!("export merged: {}", e)))?;
850850+851851+ Ok(our_root_ref.map(|root_ref| PdsEditState {
852852+ root_ref,
853853+ last_diff_ref: our_last_diff_ref,
854854+ root_snapshot: merged_snapshot.into(),
855855+ diff_updates: vec![],
856856+ last_seen_diffs: updated_last_seen,
857857+ doc_ref: merged_doc_ref.expect("Should have doc_ref if we have root"),
858858+ }))
859859+}
860860+861861+/// Remote draft info from PDS.
862862+#[derive(Clone, Debug)]
863863+pub struct RemoteDraft {
864864+ /// The draft record URI.
865865+ pub uri: AtUri<'static>,
866866+ /// The rkey (TID) of the draft.
867867+ pub rkey: String,
868868+ /// When the draft was created.
869869+ pub created_at: String,
870870+}
871871+872872+/// List all drafts for a user using weaver-index.
873873+#[cfg(feature = "use-index")]
874874+pub async fn list_drafts<C>(client: &C, did: &Did<'_>) -> Result<Vec<RemoteDraft>, CrdtError>
875875+where
876876+ C: WeaverExt,
877877+{
878878+ use jacquard::types::ident::AtIdentifier;
879879+ use weaver_api::sh_weaver::edit::list_drafts::ListDrafts;
880880+881881+ let actor = AtIdentifier::Did(did.clone().into_static());
882882+ let response = client
883883+ .send(ListDrafts::new().actor(actor).build())
884884+ .await
885885+ .map_err(|e| CrdtError::Xrpc(format!("list drafts: {}", e)))?;
886886+887887+ let output = response
888888+ .into_output()
889889+ .map_err(|e| CrdtError::Xrpc(format!("parse list drafts: {}", e)))?;
890890+891891+ tracing::debug!("list_drafts (index): found {} drafts", output.drafts.len());
892892+893893+ let drafts = output
894894+ .drafts
895895+ .into_iter()
896896+ .filter_map(|draft| {
897897+ let uri = AtUri::new(draft.uri.as_ref()).ok()?.into_static();
898898+ let rkey = uri.rkey()?.0.as_str().to_string();
899899+ let created_at = draft.created_at.to_string();
900900+ Some(RemoteDraft {
901901+ uri,
902902+ rkey,
903903+ created_at,
904904+ })
905905+ })
906906+ .collect();
907907+908908+ Ok(drafts)
909909+}
910910+911911+/// List all drafts for a user (direct PDS query, no index).
912912+#[cfg(not(feature = "use-index"))]
913913+pub async fn list_drafts<C>(client: &C, did: &Did<'_>) -> Result<Vec<RemoteDraft>, CrdtError>
914914+where
915915+ C: WeaverExt,
916916+{
917917+ use weaver_api::com_atproto::repo::list_records::ListRecords;
918918+919919+ let pds_url = client
920920+ .pds_for_did(did)
921921+ .await
922922+ .map_err(|e| CrdtError::Xrpc(format!("resolve DID: {}", e)))?;
923923+924924+ let collection =
925925+ Nsid::new(DRAFT_NSID).map_err(|e| CrdtError::InvalidUri(format!("nsid: {}", e)))?;
926926+927927+ let request = ListRecords::new()
928928+ .repo(did.clone())
929929+ .collection(collection)
930930+ .limit(100)
931931+ .build();
932932+933933+ let response = client
934934+ .xrpc(pds_url)
935935+ .send(&request)
936936+ .await
937937+ .map_err(|e| CrdtError::Xrpc(format!("list records: {}", e)))?;
938938+939939+ let output = response
940940+ .into_output()
941941+ .map_err(|e| CrdtError::Xrpc(format!("parse list records: {}", e)))?;
942942+943943+ let mut drafts = Vec::new();
944944+ for record in output.records {
945945+ let rkey = record
946946+ .uri
947947+ .rkey()
948948+ .map(|r| r.0.as_str().to_string())
949949+ .unwrap_or_default();
950950+951951+ let created_at = jacquard::from_data::<Draft>(&record.value)
952952+ .map(|d| d.created_at.to_string())
953953+ .unwrap_or_default();
954954+955955+ drafts.push(RemoteDraft {
956956+ uri: record.uri.into_static(),
957957+ rkey,
958958+ created_at,
959959+ });
960960+ }
961961+962962+ Ok(drafts)
963963+}
+11
crates/weaver-editor-crdt/src/worker/mod.rs
···11+//! Worker implementation for off-main-thread CRDT operations.
22+//!
33+//! Currently WASM-specific using gloo-worker, but the core state machine
44+//! could be abstracted to work with any async channel pair.
55+66+mod reactor;
77+88+pub use reactor::{WorkerInput, WorkerOutput};
99+1010+#[cfg(all(target_family = "wasm", target_os = "unknown"))]
1111+pub use reactor::EditorReactor;