+42

Cargo.lock

reviewed

··· 3 3 version = 4 4 4 5 5 [[package]] 6 6 + name = "adler2" 7 7 + version = "2.0.1" 8 8 + source = "registry+https://github.com/rust-lang/crates.io-index" 9 9 + checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" 10 10 + 11 11 + [[package]] 6 12 name = "android_system_properties" 7 13 version = "0.1.5" 8 14 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 246 252 checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" 247 253 248 254 [[package]] 255 255 + name = "crc32fast" 256 256 + version = "1.5.0" 257 257 + source = "registry+https://github.com/rust-lang/crates.io-index" 258 258 + checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" 259 259 + dependencies = [ 260 260 + "cfg-if", 261 261 + ] 262 262 + 263 263 + [[package]] 249 264 name = "crossbeam-utils" 250 265 version = "0.8.21" 251 266 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 338 353 checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" 339 354 340 355 [[package]] 356 356 + name = "flate2" 357 357 + version = "1.1.9" 358 358 + source = "registry+https://github.com/rust-lang/crates.io-index" 359 359 + checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" 360 360 + dependencies = [ 361 361 + "crc32fast", 362 362 + "miniz_oxide", 363 363 + ] 364 364 + 365 365 + [[package]] 341 366 name = "fnv" 342 367 version = "1.0.7" 343 368 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 836 861 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" 837 862 838 863 [[package]] 864 864 + name = "miniz_oxide" 865 865 + version = "0.8.9" 866 866 + source = "registry+https://github.com/rust-lang/crates.io-index" 867 867 + checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" 868 868 + dependencies = [ 869 869 + "adler2", 870 870 + "simd-adler32", 871 871 + ] 872 872 + 873 873 + [[package]] 839 874 name = "mio" 840 875 version = "1.1.1" 841 876 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 963 998 dependencies = [ 964 999 "chrono", 965 1000 "clap", 1001 1001 + "flate2", 966 1002 "reqwest", 967 1003 "serde", 968 1004 "serde_json", ··· 1274 1310 "errno", 1275 1311 "libc", 1276 1312 ] 1313 1313 + 1314 1314 + [[package]] 1315 1315 + name = "simd-adler32" 1316 1316 + version = "0.3.8" 1317 1317 + source = "registry+https://github.com/rust-lang/crates.io-index" 1318 1318 + checksum = "e320a6c5ad31d271ad523dcf3ad13e2767ad8b1cb8f047f75a8aeaf8da139da2" 1277 1319 1278 1320 [[package]] 1279 1321 name = "similar"

+1

Cargo.toml

reviewed

··· 20 20 tokio = { version = "1", features = ["full"] } 21 21 chrono = { version = "0.4", features = ["serde"] } 22 22 similar = "2.6" 23 23 + flate2 = "1.0" 23 24 24 25 [dev-dependencies] 25 26 tempfile = "3"

+2 -2

src/export.rs

reviewed

··· 26 26 .await? 27 27 .ok_or_else(|| format!("site record not found: {}", rkey))?; 28 28 29 29 - let site: SiteRecord = serde_json::from_value(record.value) 30 30 - .map_err(|e| format!("parse SiteRecord: {}", e))?; 29 29 + let site: SiteRecord = 30 30 + serde_json::from_value(record.value).map_err(|e| format!("parse SiteRecord: {}", e))?; 31 31 32 32 let mut files_exported = 0; 33 33

+9 -3

src/lib.rs

reviewed

··· 10 10 pub mod pack; 11 11 pub mod pds_client; 12 12 pub mod save; 13 13 + pub mod sync; 13 14 pub mod types; 14 15 pub mod yrs_pds; 15 16 ··· 17 18 pub use load::load; 18 19 pub use local_state::LocalState; 19 20 pub use merge::merge_sites; 20 20 - pub use pack::{create_pack, parse_pack, extract_entry, PackBlob, PackDataType, PackEntry}; 21 21 + pub use pack::{ 22 22 + chunk_data, compress, create_compressed_pack, create_pack, decompress, extract_entry, is_gzip, 23 23 + is_precompressed_extension, parse_pack, parse_pack_auto, reassemble_chunks, PackBlob, 24 24 + PackDataType, PackEntry, CHUNK_SIZE, 25 25 + }; 21 26 pub use pds_client::PdsClient; 22 22 - pub use save::save; 23 23 - pub use types::{COLLECTION, MANIFEST_KEY, FileKind}; 27 27 + pub use save::{save, save_filtered}; 28 28 + pub use sync::{sync_loop, SyncConfig, SyncCycleResult}; 29 29 + pub use types::{FileKind, COLLECTION, MANIFEST_KEY};

+98 -19

src/load.rs

reviewed

··· 1 1 //! Load a site from PDS into a local directory. 2 2 3 3 + use std::collections::HashMap; 3 4 use std::path::Path; 4 5 5 6 use crate::pds_client::PdsClient; ··· 10 11 /// 11 12 /// If a manifest exists, only loads files listed in the manifest. 12 13 /// Supports both text (Yrs CRDT) and binary files. 14 14 + /// Pack blobs are cached to avoid redundant downloads. 13 15 pub async fn load( 14 16 client: &PdsClient, 15 17 did: &str, ··· 23 25 .await? 24 26 .ok_or_else(|| format!("site record not found: {}", rkey))?; 25 27 26 26 - let site: SiteRecord = serde_json::from_value(record.value) 27 27 - .map_err(|e| format!("parse SiteRecord: {}", e))?; 28 28 + let site: SiteRecord = 29 29 + serde_json::from_value(record.value).map_err(|e| format!("parse SiteRecord: {}", e))?; 28 30 29 31 // Determine which files to load 30 32 let file_list = if let Some(manifest_entry) = site.files.get(MANIFEST_KEY) { 31 31 - // Manifest exists — only load listed files 32 33 let manifest_doc = yrs_pds::file_entry_to_doc(manifest_entry, client, did).await?; 33 34 let entries = yrs_pds::manifest_entries(&manifest_doc); 34 35 if verbose { ··· 36 37 } 37 38 Some(entries) 38 39 } else { 39 39 - // No manifest — load all files (legacy behavior) 40 40 None 41 41 }; 42 42 43 43 let mut files_loaded = 0; 44 44 let mut blobs_downloaded = 0; 45 45 46 46 + // Cache for pack blobs (keyed by CID) to avoid re-downloading 47 47 + let mut pack_cache: HashMap<String, Vec<u8>> = HashMap::new(); 48 48 + 46 49 for (rel_path, entry) in &site.files { 47 47 - // Skip the manifest entry itself 48 50 if rel_path == MANIFEST_KEY { 49 51 continue; 50 52 } 51 53 52 52 - // If manifest exists, only load files listed in it 53 54 if let Some(ref manifest) = file_list { 54 55 if !manifest.contains_key(rel_path) { 55 56 if verbose { ··· 71 72 72 73 match entry.kind { 73 74 FileKind::Binary => { 74 74 - // Binary file — download raw blob and write bytes 75 75 - let data = client 76 76 - .get_blob(did, entry.snapshot_blob.cid()) 77 77 - .await?; 78 78 - blobs_downloaded += 1; 75 75 + let data = get_blob_data_cached( 76 76 + entry, 77 77 + client, 78 78 + did, 79 79 + &mut pack_cache, 80 80 + &mut blobs_downloaded, 81 81 + ) 82 82 + .await?; 79 83 std::fs::write(&output_path, &data) 80 84 .map_err(|e| format!("write {:?}: {}", output_path, e))?; 81 85 } 82 86 FileKind::Text => { 83 83 - // Text file — reconstruct Yrs Doc and materialize 84 84 - let doc = yrs_pds::file_entry_to_doc(entry, client, did).await?; 85 85 - blobs_downloaded += 1; 86 86 - if entry.updates_blob.is_some() { 87 87 + // If pack_ref exists, use cached pack blob extraction 88 88 + if entry.pack_ref.is_some() { 89 89 + let snapshot_data = get_blob_data_cached( 90 90 + entry, 91 91 + client, 92 92 + did, 93 93 + &mut pack_cache, 94 94 + &mut blobs_downloaded, 95 95 + ) 96 96 + .await?; 97 97 + let doc = yrs_pds::doc_from_snapshot(&snapshot_data)?; 98 98 + if let Some(ref updates_blob) = entry.updates_blob { 99 99 + let updates_data = client.get_blob(did, updates_blob.cid()).await?; 100 100 + blobs_downloaded += 1; 101 101 + yrs_pds::apply_update(&doc, &updates_data)?; 102 102 + } 103 103 + let content = yrs_pds::materialize(&doc); 104 104 + std::fs::write(&output_path, &content) 105 105 + .map_err(|e| format!("write {:?}: {}", output_path, e))?; 106 106 + } else { 107 107 + // Direct blob download (no pack ref) 108 108 + let doc = yrs_pds::file_entry_to_doc(entry, client, did).await?; 87 109 blobs_downloaded += 1; 110 110 + if entry.updates_blob.is_some() { 111 111 + blobs_downloaded += 1; 112 112 + } 113 113 + let content = yrs_pds::materialize(&doc); 114 114 + std::fs::write(&output_path, &content) 115 115 + .map_err(|e| format!("write {:?}: {}", output_path, e))?; 88 116 } 89 89 - let content = yrs_pds::materialize(&doc); 90 90 - std::fs::write(&output_path, &content) 91 91 - .map_err(|e| format!("write {:?}: {}", output_path, e))?; 92 117 } 93 118 } 94 119 ··· 96 121 } 97 122 98 123 if verbose { 99 99 - eprintln!("pds-yrs: loaded {} file(s)", files_loaded); 124 124 + eprintln!( 125 125 + "pds-yrs: loaded {} file(s), {} blob(s) downloaded", 126 126 + files_loaded, blobs_downloaded 127 127 + ); 100 128 } 101 129 102 130 Ok(LoadResult { ··· 104 132 blobs_downloaded, 105 133 }) 106 134 } 135 135 + 136 136 + /// Get blob data for a file entry, using pack cache when available. 137 137 + async fn get_blob_data_cached( 138 138 + entry: &crate::types::FileEntry, 139 139 + client: &PdsClient, 140 140 + did: &str, 141 141 + pack_cache: &mut HashMap<String, Vec<u8>>, 142 142 + blobs_downloaded: &mut usize, 143 143 + ) -> Result<Vec<u8>, String> { 144 144 + if let Some(ref pack_ref) = entry.pack_ref { 145 145 + let cid = pack_ref.blob.cid().to_string(); 146 146 + 147 147 + // Fetch pack blob (or use cache), handling chunked packs 148 148 + if !pack_cache.contains_key(&cid) { 149 149 + let data = if let Some(ref chunks) = pack_ref.chunks { 150 150 + // Reassemble chunked pack 151 151 + let mut chunk_data = Vec::new(); 152 152 + for chunk_ref in chunks { 153 153 + let chunk = client.get_blob(did, chunk_ref.cid()).await?; 154 154 + *blobs_downloaded += 1; 155 155 + chunk_data.push(chunk); 156 156 + } 157 157 + crate::pack::reassemble_chunks(&chunk_data) 158 158 + } else { 159 159 + let d = client.get_blob(did, &cid).await?; 160 160 + *blobs_downloaded += 1; 161 161 + d 162 162 + }; 163 163 + pack_cache.insert(cid.clone(), data); 164 164 + } 165 165 + 166 166 + let pack_data = pack_cache.get(&cid).unwrap(); 167 167 + let (_, blob_data) = crate::pack::parse_pack_auto(pack_data)?; 168 168 + 169 169 + let start = pack_ref.offset as usize; 170 170 + let end = start + pack_ref.length as usize; 171 171 + if end > blob_data.len() { 172 172 + return Err(format!( 173 173 + "pack_ref out of bounds: {}..{} in {} bytes", 174 174 + start, 175 175 + end, 176 176 + blob_data.len() 177 177 + )); 178 178 + } 179 179 + Ok(blob_data[start..end].to_vec()) 180 180 + } else { 181 181 + let data = client.get_blob(did, entry.snapshot_blob.cid()).await?; 182 182 + *blobs_downloaded += 1; 183 183 + Ok(data) 184 184 + } 185 185 + }

+11 -16

src/local_state.rs

reviewed

··· 45 45 /// Save config. 46 46 pub fn save_config(&self, config: &LocalConfig) -> Result<(), String> { 47 47 let path = self.state_dir.join("_config.json"); 48 48 - let json = serde_json::to_string_pretty(config) 49 49 - .map_err(|e| format!("serialize config: {}", e))?; 50 50 - std::fs::write(&path, json) 51 51 - .map_err(|e| format!("write config: {}", e)) 48 48 + let json = 49 49 + serde_json::to_string_pretty(config).map_err(|e| format!("serialize config: {}", e))?; 50 50 + std::fs::write(&path, json).map_err(|e| format!("write config: {}", e)) 52 51 } 53 52 54 53 /// Load config. ··· 57 56 if !path.exists() { 58 57 return Ok(None); 59 58 } 60 60 - let json = std::fs::read_to_string(&path) 61 61 - .map_err(|e| format!("read config: {}", e))?; 62 62 - let config: LocalConfig = serde_json::from_str(&json) 63 63 - .map_err(|e| format!("parse config: {}", e))?; 59 59 + let json = std::fs::read_to_string(&path).map_err(|e| format!("read config: {}", e))?; 60 60 + let config: LocalConfig = 61 61 + serde_json::from_str(&json).map_err(|e| format!("parse config: {}", e))?; 64 62 Ok(Some(config)) 65 63 } 66 64 ··· 90 88 if !yrs_path.exists() { 91 89 return Ok(None); 92 90 } 93 93 - let data = std::fs::read(&yrs_path) 94 94 - .map_err(|e| format!("read doc state {}: {}", rel_path, e))?; 91 91 + let data = 92 92 + std::fs::read(&yrs_path).map_err(|e| format!("read doc state {}: {}", rel_path, e))?; 95 93 let doc = yrs_pds::doc_from_snapshot(&data)?; 96 94 Ok(Some(doc)) 97 95 } ··· 118 116 if !yrs_path.exists() { 119 117 return Ok(None); 120 118 } 121 121 - let data = std::fs::read(&yrs_path) 122 122 - .map_err(|e| format!("read manifest state: {}", e))?; 119 119 + let data = std::fs::read(&yrs_path).map_err(|e| format!("read manifest state: {}", e))?; 123 120 let doc = yrs_pds::manifest_from_snapshot(&data)?; 124 121 Ok(Some(doc)) 125 122 } ··· 153 150 std::fs::create_dir_all(parent) 154 151 .map_err(|e| format!("create dir for cid {}: {}", rel_path, e))?; 155 152 } 156 156 - std::fs::write(&path, cid) 157 157 - .map_err(|e| format!("write snapshot cid {}: {}", rel_path, e)) 153 153 + std::fs::write(&path, cid).map_err(|e| format!("write snapshot cid {}: {}", rel_path, e)) 158 154 } 159 155 160 156 /// Load the last-seen snapshot blob CID for a file. ··· 186 182 dir: &Path, 187 183 files: &mut Vec<String>, 188 184 ) -> Result<(), String> { 189 189 - let entries = std::fs::read_dir(dir) 190 190 - .map_err(|e| format!("read dir {:?}: {}", dir, e))?; 185 185 + let entries = std::fs::read_dir(dir).map_err(|e| format!("read dir {:?}: {}", dir, e))?; 191 186 for entry in entries.flatten() { 192 187 let path = entry.path(); 193 188 if path.is_dir() {

+126 -27

src/main.rs

reviewed

··· 2 2 use std::process; 3 3 4 4 #[derive(Parser)] 5 5 - #[command(name = "pds-yrs", about = "Sync Yrs CRDT documents via AT Protocol PDS")] 5 5 + #[command( 6 6 + name = "pds-yrs", 7 7 + about = "Sync Yrs CRDT documents via AT Protocol PDS" 8 8 + )] 6 9 struct Cli { 7 10 #[command(subcommand)] 8 11 command: Command, ··· 27 30 /// PDS URL (default: resolved from handle) 28 31 #[arg(long)] 29 32 pds: Option<String>, 33 33 + /// Include glob patterns (e.g. "*.md") 34 34 + #[arg(long)] 35 35 + include: Vec<String>, 36 36 + /// Exclude glob patterns (e.g. "*.tmp") 37 37 + #[arg(long)] 38 38 + exclude: Vec<String>, 30 39 /// Show progress 31 40 #[arg(long)] 32 41 verbose: bool, ··· 73 82 #[arg(long)] 74 83 verbose: bool, 75 84 }, 85 85 + /// Sync directory with PDS in real-time (WebRTC fallback) 86 86 + Sync { 87 87 + /// Directory to sync 88 88 + #[arg(long)] 89 89 + dir: String, 90 90 + /// AT Protocol handle 91 91 + #[arg(long)] 92 92 + handle: String, 93 93 + /// Site name (used as rkey) 94 94 + #[arg(long)] 95 95 + site: String, 96 96 + /// Password for authentication 97 97 + #[arg(long)] 98 98 + password: String, 99 99 + /// PDS URL 100 100 + #[arg(long)] 101 101 + pds: Option<String>, 102 102 + /// Poll interval in seconds 103 103 + #[arg(long, default_value = "3")] 104 104 + interval: u64, 105 105 + /// Include glob patterns (e.g. "*.md") 106 106 + #[arg(long)] 107 107 + include: Vec<String>, 108 108 + /// Exclude glob patterns (e.g. "*.tmp") 109 109 + #[arg(long)] 110 110 + exclude: Vec<String>, 111 111 + /// Show progress 112 112 + #[arg(long)] 113 113 + verbose: bool, 114 114 + }, 76 115 /// Export site content as plain text (no Yrs decoding needed) 77 116 Export { 78 117 /// AT Protocol handle ··· 107 146 site, 108 147 password, 109 148 pds, 149 149 + include, 150 150 + exclude, 110 151 verbose, 111 111 - } => run_save(&dir, &handle, &site, &password, pds.as_deref(), verbose).await, 152 152 + } => { 153 153 + run_save( 154 154 + &dir, 155 155 + &handle, 156 156 + &site, 157 157 + &password, 158 158 + pds.as_deref(), 159 159 + &include, 160 160 + &exclude, 161 161 + verbose, 162 162 + ) 163 163 + .await 164 164 + } 112 165 Command::Load { 113 166 handle, 114 167 site, ··· 125 178 pds, 126 179 verbose, 127 180 } => run_merge(&sites, &handle, &output, &password, pds.as_deref(), verbose).await, 181 181 + Command::Sync { 182 182 + dir, 183 183 + handle, 184 184 + site, 185 185 + password, 186 186 + pds, 187 187 + interval, 188 188 + include, 189 189 + exclude, 190 190 + verbose, 191 191 + } => { 192 192 + run_sync( 193 193 + &dir, 194 194 + &handle, 195 195 + &site, 196 196 + &password, 197 197 + pds.as_deref(), 198 198 + interval, 199 199 + &include, 200 200 + &exclude, 201 201 + verbose, 202 202 + ) 203 203 + .await 204 204 + } 128 205 Command::Export { 129 206 handle, 130 207 site, ··· 158 235 site: &str, 159 236 password: &str, 160 237 pds_url: Option<&str>, 238 238 + include: &[String], 239 239 + exclude: &[String], 161 240 verbose: bool, 162 241 ) -> Result<(), String> { 163 242 let (client, did) = login(handle, password, pds_url).await?; 164 164 - let result = pds_yrs::save( 243 243 + let inc = if include.is_empty() { 244 244 + None 245 245 + } else { 246 246 + Some(include) 247 247 + }; 248 248 + let exc = if exclude.is_empty() { 249 249 + None 250 250 + } else { 251 251 + Some(exclude) 252 252 + }; 253 253 + let result = pds_yrs::save_filtered( 165 254 std::path::Path::new(dir), 166 255 &client, 167 256 &did, 168 257 site, 258 258 + inc, 259 259 + exc, 169 260 verbose, 170 261 ) 171 262 .await?; ··· 185 276 verbose: bool, 186 277 ) -> Result<(), String> { 187 278 let (client, did) = login(handle, password, pds_url).await?; 188 188 - let result = pds_yrs::load( 189 189 - &client, 190 190 - &did, 191 191 - site, 192 192 - std::path::Path::new(output), 193 193 - verbose, 194 194 - ) 195 195 - .await?; 279 279 + let result = pds_yrs::load(&client, &did, site, std::path::Path::new(output), verbose).await?; 196 280 eprintln!( 197 281 "pds-yrs: loaded {} file(s), {} blob(s) downloaded", 198 282 result.files_loaded, result.blobs_downloaded ··· 210 294 ) -> Result<(), String> { 211 295 let (client, did) = login(handle, password, pds_url).await?; 212 296 let rkeys: Vec<&str> = sites.split(',').collect(); 213 213 - pds_yrs::merge_sites( 214 214 - &client, 215 215 - &did, 216 216 - &rkeys, 217 217 - std::path::Path::new(output), 218 218 - verbose, 219 219 - ) 220 220 - .await?; 297 297 + pds_yrs::merge_sites(&client, &did, &rkeys, std::path::Path::new(output), verbose).await?; 221 298 eprintln!("pds-yrs: merge complete"); 222 299 Ok(()) 223 300 } 224 301 302 302 + async fn run_sync( 303 303 + dir: &str, 304 304 + handle: &str, 305 305 + site: &str, 306 306 + password: &str, 307 307 + pds_url: Option<&str>, 308 308 + interval: u64, 309 309 + include: &[String], 310 310 + exclude: &[String], 311 311 + verbose: bool, 312 312 + ) -> Result<(), String> { 313 313 + let (client, did) = login(handle, password, pds_url).await?; 314 314 + let config = pds_yrs::SyncConfig { 315 315 + dir: dir.to_string(), 316 316 + interval: std::time::Duration::from_secs(interval), 317 317 + max_cycles: None, 318 318 + materialize_interval: 0, 319 319 + include: include.to_vec(), 320 320 + exclude: exclude.to_vec(), 321 321 + verbose, 322 322 + }; 323 323 + eprintln!( 324 324 + "pds-yrs: starting sync (interval={}s, Ctrl+C to stop)", 325 325 + interval 326 326 + ); 327 327 + pds_yrs::sync_loop(&client, &did, site, &config).await?; 328 328 + Ok(()) 329 329 + } 330 330 + 225 331 async fn run_export( 226 332 handle: &str, 227 333 site: &str, ··· 231 337 verbose: bool, 232 338 ) -> Result<(), String> { 233 339 let (client, did) = login(handle, password, pds_url).await?; 234 234 - let count = pds_yrs::export( 235 235 - &client, 236 236 - &did, 237 237 - site, 238 238 - std::path::Path::new(output), 239 239 - verbose, 240 240 - ) 241 241 - .await?; 340 340 + let count = pds_yrs::export(&client, &did, site, std::path::Path::new(output), verbose).await?; 242 341 eprintln!("pds-yrs: exported {} file(s)", count); 243 342 Ok(()) 244 343 }

+152 -5

src/merge.rs

reviewed

··· 40 40 let manifest_entries = yrs_pds::manifest_entries(&merged_manifest); 41 41 42 42 if verbose { 43 43 - eprintln!("pds-yrs: merged manifest has {} file(s)", manifest_entries.len()); 43 43 + eprintln!( 44 44 + "pds-yrs: merged manifest has {} file(s)", 45 45 + manifest_entries.len() 46 46 + ); 44 47 } 45 48 46 49 // For each file in the merged manifest, merge across sites ··· 61 64 eprintln!( 62 65 "pds-yrs: merging {} ({}, from {} site(s))", 63 66 rel_path, 64 64 - match kind { FileKind::Text => "text", FileKind::Binary => "binary" }, 67 67 + match kind { 68 68 + FileKind::Text => "text", 69 69 + FileKind::Binary => "binary", 70 70 + }, 65 71 site_indices.len() 66 72 ); 67 73 } ··· 79 85 .map_err(|e| format!("write {:?}: {}", output_path, e))?; 80 86 } 81 87 FileKind::Binary => { 82 82 - merge_binary_file(rel_path, &site_indices, &sites, rkeys, client, did, output_dir).await?; 88 88 + merge_binary_file( 89 89 + rel_path, 90 90 + &site_indices, 91 91 + &sites, 92 92 + rkeys, 93 93 + client, 94 94 + did, 95 95 + output_dir, 96 96 + ) 97 97 + .await?; 83 98 } 84 99 } 85 100 } ··· 219 234 }; 220 235 let conflict_path = output_dir.join(parent).join(&conflict_name); 221 236 if let Some(p) = conflict_path.parent() { 222 222 - std::fs::create_dir_all(p) 223 223 - .map_err(|e| format!("create dir {:?}: {}", p, e))?; 237 237 + std::fs::create_dir_all(p).map_err(|e| format!("create dir {:?}: {}", p, e))?; 224 238 } 225 239 let data = client.get_blob(did, cid).await?; 226 240 std::fs::write(&conflict_path, &data) ··· 230 244 231 245 Ok(()) 232 246 } 247 247 + 248 248 + /// Generate a conflict filename: stem.site_name.ext 249 249 + pub fn conflict_filename(rel_path: &str, site_name: &str) -> String { 250 250 + let path = std::path::Path::new(rel_path); 251 251 + let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("file"); 252 252 + let ext = path.extension().and_then(|s| s.to_str()).unwrap_or(""); 253 253 + let parent = path.parent().unwrap_or(std::path::Path::new("")); 254 254 + 255 255 + let conflict_name = if ext.is_empty() { 256 256 + format!("{}.{}", stem, site_name) 257 257 + } else { 258 258 + format!("{}.{}.{}", stem, site_name, ext) 259 259 + }; 260 260 + 261 261 + parent.join(&conflict_name).to_string_lossy().to_string() 262 262 + } 263 263 + 264 264 + #[cfg(test)] 265 265 + mod tests { 266 266 + use super::*; 267 267 + 268 268 + #[test] 269 269 + fn conflict_filename_with_extension() { 270 270 + assert_eq!(conflict_filename("logo.png", "site-a"), "logo.site-a.png"); 271 271 + } 272 272 + 273 273 + #[test] 274 274 + fn conflict_filename_without_extension() { 275 275 + assert_eq!(conflict_filename("Makefile", "site-b"), "Makefile.site-b"); 276 276 + } 277 277 + 278 278 + #[test] 279 279 + fn conflict_filename_nested_path() { 280 280 + assert_eq!( 281 281 + conflict_filename("images/photo.jpg", "collab1"), 282 282 + "images/photo.collab1.jpg" 283 283 + ); 284 284 + } 285 285 + 286 286 + #[test] 287 287 + fn binary_cid_comparison_same() { 288 288 + let mut cid_map: HashMap<String, Vec<usize>> = HashMap::new(); 289 289 + cid_map.entry("bafyabc123".to_string()).or_default().push(0); 290 290 + cid_map.entry("bafyabc123".to_string()).or_default().push(1); 291 291 + assert_eq!(cid_map.len(), 1, "same CID should produce no conflict"); 292 292 + } 293 293 + 294 294 + #[test] 295 295 + fn binary_cid_comparison_different() { 296 296 + let mut cid_map: HashMap<String, Vec<usize>> = HashMap::new(); 297 297 + cid_map.entry("bafyabc123".to_string()).or_default().push(0); 298 298 + cid_map.entry("bafydef456".to_string()).or_default().push(1); 299 299 + assert_eq!(cid_map.len(), 2, "different CIDs should produce conflict"); 300 300 + } 301 301 + 302 302 + #[test] 303 303 + fn conflict_filename_preserves_deeply_nested() { 304 304 + assert_eq!( 305 305 + conflict_filename("a/b/c/deep.png", "alice"), 306 306 + "a/b/c/deep.alice.png" 307 307 + ); 308 308 + } 309 309 + 310 310 + #[test] 311 311 + fn binary_merge_no_conflict_same_cid() { 312 312 + // simulate: two sites have same binary CID → no conflict 313 313 + let mut cid_site: HashMap<String, Vec<usize>> = HashMap::new(); 314 314 + let cid = "bafysame".to_string(); 315 315 + cid_site.entry(cid.clone()).or_default().push(0); 316 316 + cid_site.entry(cid).or_default().push(1); 317 317 + 318 318 + // single entry in map → no conflict 319 319 + assert_eq!(cid_site.len(), 1); 320 320 + } 321 321 + 322 322 + #[test] 323 323 + fn binary_merge_conflict_different_cids() { 324 324 + // simulate: two sites modify same binary differently → conflict files 325 325 + let mut cid_site: HashMap<String, Vec<usize>> = HashMap::new(); 326 326 + cid_site.entry("bafyabc".to_string()).or_default().push(0); 327 327 + cid_site.entry("bafydef".to_string()).or_default().push(1); 328 328 + 329 329 + assert_eq!(cid_site.len(), 2); 330 330 + 331 331 + // verify conflict filenames 332 332 + let rkeys = ["site-a", "site-b"]; 333 333 + let mut conflict_files = Vec::new(); 334 334 + for (_, indices) in &cid_site { 335 335 + let site_name = rkeys[indices[0]]; 336 336 + conflict_files.push(conflict_filename("logo.png", site_name)); 337 337 + } 338 338 + conflict_files.sort(); 339 339 + assert!(conflict_files.contains(&"logo.site-a.png".to_string())); 340 340 + assert!(conflict_files.contains(&"logo.site-b.png".to_string())); 341 341 + } 342 342 + 343 343 + #[test] 344 344 + fn text_crdt_merge_concurrent_edits_same_doc() { 345 345 + use yrs::updates::decoder::Decode; 346 346 + use yrs::{Text, Transact}; 347 347 + 348 348 + // Simulate two sites editing the same text file 349 349 + let base_doc = crate::yrs_pds::doc_from_text("Hello world"); 350 350 + let base_snapshot = crate::yrs_pds::encode_snapshot(&base_doc); 351 351 + 352 352 + // Site A: adds " from Alice" at end 353 353 + let doc_a = crate::yrs_pds::doc_from_snapshot(&base_snapshot).unwrap(); 354 354 + { 355 355 + let text = doc_a.get_or_insert_text("content"); 356 356 + let mut txn = doc_a.transact_mut(); 357 357 + text.insert(&mut txn, 11, " from Alice"); 358 358 + } 359 359 + 360 360 + // Site B: adds "Dear " at beginning 361 361 + let doc_b = crate::yrs_pds::doc_from_snapshot(&base_snapshot).unwrap(); 362 362 + { 363 363 + let text = doc_b.get_or_insert_text("content"); 364 364 + let mut txn = doc_b.transact_mut(); 365 365 + text.insert(&mut txn, 0, "Dear "); 366 366 + } 367 367 + 368 368 + // Merge B into A 369 369 + let sv_a = doc_a.transact().state_vector(); 370 370 + let diff_b = doc_b.transact().encode_diff_v1(&sv_a); 371 371 + let update = yrs::Update::decode_v1(&diff_b).unwrap(); 372 372 + doc_a.transact_mut().apply_update(update).unwrap(); 373 373 + 374 374 + let merged = crate::yrs_pds::materialize(&doc_a); 375 375 + // CRDT should include both edits 376 376 + assert!(merged.contains("Dear "), "should have site B's prefix"); 377 377 + assert!(merged.contains("from Alice"), "should have site A's suffix"); 378 378 + } 379 379 + }

+310 -5

src/pack.rs

reviewed

··· 116 116 Ok(&blob_data[start..end]) 117 117 } 118 118 119 119 + /// Compress data with gzip. 120 120 + pub fn compress(data: &[u8]) -> Vec<u8> { 121 121 + use flate2::write::GzEncoder; 122 122 + use flate2::Compression; 123 123 + use std::io::Write; 124 124 + 125 125 + let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); 126 126 + encoder.write_all(data).unwrap(); 127 127 + encoder.finish().unwrap() 128 128 + } 129 129 + 130 130 + /// Decompress gzip data. 131 131 + pub fn decompress(data: &[u8]) -> Result<Vec<u8>, String> { 132 132 + use flate2::read::GzDecoder; 133 133 + use std::io::Read; 134 134 + 135 135 + let mut decoder = GzDecoder::new(data); 136 136 + let mut result = Vec::new(); 137 137 + decoder 138 138 + .read_to_end(&mut result) 139 139 + .map_err(|e| format!("decompress: {}", e))?; 140 140 + Ok(result) 141 141 + } 142 142 + 143 143 + /// Check if data looks like gzip (magic bytes 0x1f 0x8b). 144 144 + pub fn is_gzip(data: &[u8]) -> bool { 145 145 + data.len() >= 2 && data[0] == 0x1f && data[1] == 0x8b 146 146 + } 147 147 + 148 148 + /// Maximum blob size before chunking (40MB — AT Protocol limit is ~50MB but leave margin). 149 149 + pub const CHUNK_SIZE: usize = 40 * 1024 * 1024; 150 150 + 151 151 + /// Split data into chunks of at most CHUNK_SIZE bytes. 152 152 + pub fn chunk_data(data: &[u8]) -> Vec<Vec<u8>> { 153 153 + if data.len() <= CHUNK_SIZE { 154 154 + return vec![data.to_vec()]; 155 155 + } 156 156 + data.chunks(CHUNK_SIZE) 157 157 + .map(|chunk| chunk.to_vec()) 158 158 + .collect() 159 159 + } 160 160 + 161 161 + /// Reassemble chunks into a single blob. 162 162 + pub fn reassemble_chunks(chunks: &[Vec<u8>]) -> Vec<u8> { 163 163 + let total: usize = chunks.iter().map(|c| c.len()).sum(); 164 164 + let mut result = Vec::with_capacity(total); 165 165 + for chunk in chunks { 166 166 + result.extend_from_slice(chunk); 167 167 + } 168 168 + result 169 169 + } 170 170 + 171 171 + /// Check if a file extension is already compressed (skip compression). 172 172 + pub fn is_precompressed_extension(ext: &str) -> bool { 173 173 + matches!( 174 174 + ext.to_lowercase().as_str(), 175 175 + "jpg" 176 176 + | "jpeg" 177 177 + | "png" 178 178 + | "gif" 179 179 + | "webp" 180 180 + | "avif" 181 181 + | "mp4" 182 182 + | "webm" 183 183 + | "mp3" 184 184 + | "ogg" 185 185 + | "flac" 186 186 + | "zip" 187 187 + | "gz" 188 188 + | "bz2" 189 189 + | "xz" 190 190 + | "zst" 191 191 + | "br" 192 192 + | "woff2" 193 193 + | "woff" 194 194 + ) 195 195 + } 196 196 + 197 197 + /// Create a compressed pack blob. Compresses the entire pack with gzip. 198 198 + pub fn create_compressed_pack(items: &[(&str, &[u8], PackDataType)]) -> PackBlob { 199 199 + let pack = create_pack(items); 200 200 + let compressed = compress(&pack.data); 201 201 + 202 202 + // Only use compression if it actually saves space 203 203 + if compressed.len() < pack.data.len() { 204 204 + PackBlob { 205 205 + data: compressed, 206 206 + entries: pack.entries, 207 207 + } 208 208 + } else { 209 209 + pack 210 210 + } 211 211 + } 212 212 + 213 213 + /// Parse a pack blob, auto-detecting gzip compression. 214 214 + pub fn parse_pack_auto(data: &[u8]) -> Result<(Vec<PackEntry>, Vec<u8>), String> { 215 215 + let decompressed; 216 216 + let actual_data = if is_gzip(data) { 217 217 + decompressed = decompress(data)?; 218 218 + &decompressed 219 219 + } else { 220 220 + data 221 221 + }; 222 222 + 223 223 + let (entries, blob_data) = parse_pack(actual_data)?; 224 224 + Ok((entries, blob_data.to_vec())) 225 225 + } 226 226 + 119 227 #[cfg(test)] 120 228 mod tests { 121 229 use super::*; ··· 123 231 #[test] 124 232 fn pack_round_trip() { 125 233 let items: Vec<(&str, &[u8], PackDataType)> = vec![ 126 126 - ("docs/index.md", b"snapshot data for index", PackDataType::Snapshot), 127 127 - ("docs/about.md", b"snapshot data for about", PackDataType::Snapshot), 234 234 + ( 235 235 + "docs/index.md", 236 236 + b"snapshot data for index", 237 237 + PackDataType::Snapshot, 238 238 + ), 239 239 + ( 240 240 + "docs/about.md", 241 241 + b"snapshot data for about", 242 242 + PackDataType::Snapshot, 243 243 + ), 128 244 ("images/logo.png", b"raw png bytes", PackDataType::Binary), 129 245 ]; 130 246 ··· 158 274 #[test] 159 275 fn pack_single_large_entry() { 160 276 let big_data = vec![42u8; 100_000]; 161 161 - let items: Vec<(&str, &[u8], PackDataType)> = vec![ 162 162 - ("big.bin", &big_data, PackDataType::Binary), 163 163 - ]; 277 277 + let items: Vec<(&str, &[u8], PackDataType)> = 278 278 + vec![("big.bin", &big_data, PackDataType::Binary)]; 164 279 let pack = create_pack(&items); 165 280 let (entries, blob_data) = parse_pack(&pack.data).unwrap(); 166 281 let extracted = extract_entry(&entries[0], blob_data).unwrap(); ··· 171 286 #[test] 172 287 fn pack_parse_error_on_truncated() { 173 288 assert!(parse_pack(b"ab").is_err()); 289 289 + } 290 290 + 291 291 + #[test] 292 292 + fn compress_decompress_round_trip() { 293 293 + let data = b"Hello world! This is some test data for compression."; 294 294 + let compressed = compress(data); 295 295 + assert!(is_gzip(&compressed)); 296 296 + let decompressed = decompress(&compressed).unwrap(); 297 297 + assert_eq!(decompressed, data); 298 298 + } 299 299 + 300 300 + #[test] 301 301 + fn compressed_pack_round_trip() { 302 302 + // Use repetitive data so compression actually helps 303 303 + let text = "Hello world! ".repeat(100); 304 304 + let items: Vec<(&str, &[u8], PackDataType)> = 305 305 + vec![("file.md", text.as_bytes(), PackDataType::Snapshot)]; 306 306 + let pack = create_compressed_pack(&items); 307 307 + // Compressed should be smaller for repetitive data 308 308 + let uncompressed = create_pack(&items); 309 309 + assert!(pack.data.len() < uncompressed.data.len()); 310 310 + 311 311 + // Parse with auto-detection 312 312 + let (entries, blob_data) = parse_pack_auto(&pack.data).unwrap(); 313 313 + assert_eq!(entries.len(), 1); 314 314 + let extracted = extract_entry(&entries[0], &blob_data).unwrap(); 315 315 + assert_eq!(extracted, text.as_bytes()); 316 316 + } 317 317 + 318 318 + #[test] 319 319 + fn uncompressed_pack_auto_parse() { 320 320 + let items: Vec<(&str, &[u8], PackDataType)> = 321 321 + vec![("file.md", b"short", PackDataType::Snapshot)]; 322 322 + let pack = create_pack(&items); 323 323 + // Should work fine with parse_pack_auto even without compression 324 324 + let (entries, blob_data) = parse_pack_auto(&pack.data).unwrap(); 325 325 + assert_eq!(entries.len(), 1); 326 326 + let extracted = extract_entry(&entries[0], &blob_data).unwrap(); 327 327 + assert_eq!(extracted, b"short"); 328 328 + } 329 329 + 330 330 + #[test] 331 331 + fn chunk_small_data_no_split() { 332 332 + let data = vec![42u8; 100]; 333 333 + let chunks = chunk_data(&data); 334 334 + assert_eq!(chunks.len(), 1); 335 335 + assert_eq!(chunks[0], data); 336 336 + } 337 337 + 338 338 + #[test] 339 339 + fn chunk_and_reassemble() { 340 340 + // Create data larger than 1 chunk (use a small chunk size via direct splitting) 341 341 + let data: Vec<u8> = (0..256).map(|i| (i % 256) as u8).collect(); 342 342 + let chunks: Vec<Vec<u8>> = data.chunks(100).map(|c| c.to_vec()).collect(); 343 343 + assert_eq!(chunks.len(), 3); // 100 + 100 + 56 344 344 + let reassembled = reassemble_chunks(&chunks); 345 345 + assert_eq!(reassembled, data); 346 346 + } 347 347 + 348 348 + #[test] 349 349 + fn precompressed_extension_detection() { 350 350 + assert!(is_precompressed_extension("jpg")); 351 351 + assert!(is_precompressed_extension("PNG")); 352 352 + assert!(is_precompressed_extension("gz")); 353 353 + assert!(!is_precompressed_extension("md")); 354 354 + assert!(!is_precompressed_extension("html")); 355 355 + } 356 356 + 357 357 + #[test] 358 358 + fn pack_mixed_data_types() { 359 359 + // pack containing snapshot, update, and binary entries 360 360 + let items: Vec<(&str, &[u8], PackDataType)> = vec![ 361 361 + ("index.md", b"yrs snapshot bytes", PackDataType::Snapshot), 362 362 + ("index.md.update", b"yrs update diff", PackDataType::Update), 363 363 + ("logo.png", b"\x89PNG raw data", PackDataType::Binary), 364 364 + ]; 365 365 + let pack = create_pack(&items); 366 366 + let (entries, blob_data) = parse_pack(&pack.data).unwrap(); 367 367 + 368 368 + assert_eq!(entries.len(), 3); 369 369 + assert!(matches!(entries[0].data_type, PackDataType::Snapshot)); 370 370 + assert!(matches!(entries[1].data_type, PackDataType::Update)); 371 371 + assert!(matches!(entries[2].data_type, PackDataType::Binary)); 372 372 + 373 373 + assert_eq!( 374 374 + extract_entry(&entries[0], blob_data).unwrap(), 375 375 + b"yrs snapshot bytes" 376 376 + ); 377 377 + assert_eq!( 378 378 + extract_entry(&entries[1], blob_data).unwrap(), 379 379 + b"yrs update diff" 380 380 + ); 381 381 + assert_eq!( 382 382 + extract_entry(&entries[2], blob_data).unwrap(), 383 383 + b"\x89PNG raw data" 384 384 + ); 385 385 + } 386 386 + 387 387 + #[test] 388 388 + fn compress_already_compressed_skips() { 389 389 + // very short data — compression may not help 390 390 + let data = b"tiny"; 391 391 + let compressed = compress(data); 392 392 + // gzip adds overhead for tiny data 393 393 + assert!(is_gzip(&compressed)); 394 394 + let decompressed = decompress(&compressed).unwrap(); 395 395 + assert_eq!(decompressed, data); 396 396 + } 397 397 + 398 398 + #[test] 399 399 + fn compressed_pack_skips_when_no_savings() { 400 400 + // Short random-ish data where compression doesn't help 401 401 + let items: Vec<(&str, &[u8], PackDataType)> = vec![("a.bin", b"x", PackDataType::Binary)]; 402 402 + let compressed = create_compressed_pack(&items); 403 403 + let uncompressed = create_pack(&items); 404 404 + // For tiny data, create_compressed_pack should fall back to uncompressed 405 405 + // (or use compressed if smaller — either way, parse_pack_auto handles both) 406 406 + let (entries, blob_data) = parse_pack_auto(&compressed.data).unwrap(); 407 407 + assert_eq!(entries.len(), 1); 408 408 + assert_eq!(extract_entry(&entries[0], &blob_data).unwrap(), b"x"); 409 409 + // Just verify uncompressed also works 410 410 + let (entries2, blob_data2) = parse_pack_auto(&uncompressed.data).unwrap(); 411 411 + assert_eq!(extract_entry(&entries2[0], &blob_data2).unwrap(), b"x"); 412 412 + } 413 413 + 414 414 + #[test] 415 415 + fn chunk_data_boundary() { 416 416 + // data exactly at chunk size should not be split 417 417 + let data = vec![42u8; CHUNK_SIZE]; 418 418 + let chunks = chunk_data(&data); 419 419 + assert_eq!(chunks.len(), 1); 420 420 + assert_eq!(chunks[0].len(), CHUNK_SIZE); 421 421 + } 422 422 + 423 423 + #[test] 424 424 + fn chunk_data_just_over_boundary() { 425 425 + // data just over chunk size should be split into 2 426 426 + let data = vec![42u8; CHUNK_SIZE + 1]; 427 427 + let chunks = chunk_data(&data); 428 428 + assert_eq!(chunks.len(), 2); 429 429 + assert_eq!(chunks[0].len(), CHUNK_SIZE); 430 430 + assert_eq!(chunks[1].len(), 1); 431 431 + 432 432 + let reassembled = reassemble_chunks(&chunks); 433 433 + assert_eq!(reassembled, data); 434 434 + } 435 435 + 436 436 + #[test] 437 437 + fn is_gzip_false_for_non_gzip() { 438 438 + assert!(!is_gzip(b"not gzip")); 439 439 + assert!(!is_gzip(b"")); 440 440 + assert!(!is_gzip(b"\x1f")); 441 441 + assert!(!is_gzip(b"\x00\x00")); 442 442 + } 443 443 + 444 444 + #[test] 445 445 + fn pack_ref_extraction_exact_offsets() { 446 446 + // verify offset/length slicing produces correct data 447 447 + let items: Vec<(&str, &[u8], PackDataType)> = vec![ 448 448 + ("a.txt", b"AAAA", PackDataType::Snapshot), 449 449 + ("b.txt", b"BBBBBB", PackDataType::Snapshot), 450 450 + ("c.txt", b"CC", PackDataType::Snapshot), 451 451 + ]; 452 452 + let pack = create_pack(&items); 453 453 + let (entries, blob_data) = parse_pack(&pack.data).unwrap(); 454 454 + 455 455 + // check offsets are correct 456 456 + assert_eq!(entries[0].offset, 0); 457 457 + assert_eq!(entries[0].length, 4); 458 458 + assert_eq!(entries[1].offset, 4); 459 459 + assert_eq!(entries[1].length, 6); 460 460 + assert_eq!(entries[2].offset, 10); 461 461 + assert_eq!(entries[2].length, 2); 462 462 + 463 463 + // manual slice verification 464 464 + let start = entries[1].offset as usize; 465 465 + let end = start + entries[1].length as usize; 466 466 + assert_eq!(&blob_data[start..end], b"BBBBBB"); 467 467 + } 468 468 + 469 469 + #[test] 470 470 + fn extract_entry_out_of_bounds() { 471 471 + let entry = PackEntry { 472 472 + path: "bad.bin".to_string(), 473 473 + offset: 100, 474 474 + length: 50, 475 475 + data_type: PackDataType::Binary, 476 476 + }; 477 477 + let blob_data = b"short"; 478 478 + assert!(extract_entry(&entry, blob_data).is_err()); 174 479 } 175 480 }

+61 -21

src/pds_client.rs

reviewed

··· 9 9 pub struct PdsClient { 10 10 base_url: String, 11 11 auth_token: Option<String>, 12 12 + refresh_token: Option<String>, 13 13 + token_expiry: Option<std::time::Instant>, 12 14 http: reqwest::Client, 13 15 } 16 16 + 17 17 + /// Conservative token TTL — refresh before 90 minutes. 18 18 + const TOKEN_TTL_SECS: u64 = 90 * 60; 14 19 15 20 #[derive(Debug, Deserialize)] 16 21 pub struct CreateSessionResponse { ··· 66 71 Self { 67 72 base_url: base_url.into().trim_end_matches('/').to_string(), 68 73 auth_token: None, 74 74 + refresh_token: None, 75 75 + token_expiry: None, 69 76 http: reqwest::Client::new(), 70 77 } 71 78 } ··· 105 112 .map_err(|e| format!("parse login response: {}", e))?; 106 113 107 114 self.auth_token = Some(session.access_jwt.clone()); 115 115 + self.refresh_token = Some(session.refresh_jwt.clone()); 116 116 + self.token_expiry = 117 117 + Some(std::time::Instant::now() + std::time::Duration::from_secs(TOKEN_TTL_SECS)); 108 118 Ok(session) 109 119 } 110 120 121 121 + /// Refresh the access token using the refresh token. 122 122 + pub async fn refresh_session(&mut self) -> Result<(), String> { 123 123 + let refresh_jwt = self 124 124 + .refresh_token 125 125 + .as_ref() 126 126 + .ok_or("no refresh token")? 127 127 + .clone(); 128 128 + 129 129 + let url = format!("{}/xrpc/com.atproto.server.refreshSession", self.base_url); 130 130 + 131 131 + let resp = self 132 132 + .http 133 133 + .post(&url) 134 134 + .bearer_auth(&refresh_jwt) 135 135 + .send() 136 136 + .await 137 137 + .map_err(|e| format!("refresh request failed: {}", e))?; 138 138 + 139 139 + if !resp.status().is_success() { 140 140 + let status = resp.status(); 141 141 + let text = resp.text().await.unwrap_or_default(); 142 142 + return Err(format!("refresh failed ({}): {}", status, text)); 143 143 + } 144 144 + 145 145 + let session: CreateSessionResponse = resp 146 146 + .json() 147 147 + .await 148 148 + .map_err(|e| format!("parse refresh response: {}", e))?; 149 149 + 150 150 + self.auth_token = Some(session.access_jwt); 151 151 + self.refresh_token = Some(session.refresh_jwt); 152 152 + self.token_expiry = 153 153 + Some(std::time::Instant::now() + std::time::Duration::from_secs(TOKEN_TTL_SECS)); 154 154 + Ok(()) 155 155 + } 156 156 + 157 157 + /// Ensure auth token is fresh — refresh if near expiry. 158 158 + pub async fn ensure_fresh_token(&mut self) -> Result<(), String> { 159 159 + if let Some(expiry) = self.token_expiry { 160 160 + if std::time::Instant::now() >= expiry { 161 161 + self.refresh_session().await?; 162 162 + } 163 163 + } 164 164 + Ok(()) 165 165 + } 166 166 + 111 167 pub async fn get_record( 112 168 &self, 113 169 did: &str, ··· 156 212 swap_record: Option<String>, 157 213 ) -> Result<PutRecordResponse, String> { 158 214 let url = format!("{}/xrpc/com.atproto.repo.putRecord", self.base_url); 159 159 - let token = self 160 160 - .auth_token 161 161 - .as_ref() 162 162 - .ok_or("not authenticated")?; 215 215 + let token = self.auth_token.as_ref().ok_or("not authenticated")?; 163 216 164 217 let body = PutRecordRequest { 165 218 repo: did.to_string(), ··· 191 244 192 245 pub async fn upload_blob(&self, data: Vec<u8>) -> Result<BlobRef, String> { 193 246 let url = format!("{}/xrpc/com.atproto.repo.uploadBlob", self.base_url); 194 194 - let token = self 195 195 - .auth_token 196 196 - .as_ref() 197 197 - .ok_or("not authenticated")?; 247 247 + let token = self.auth_token.as_ref().ok_or("not authenticated")?; 198 248 199 249 let resp = self 200 250 .http ··· 250 300 .map_err(|e| format!("read blob bytes: {}", e)) 251 301 } 252 302 253 253 - pub async fn delete_record( 254 254 - &self, 255 255 - collection: &str, 256 256 - rkey: &str, 257 257 - ) -> Result<(), String> { 303 303 + pub async fn delete_record(&self, collection: &str, rkey: &str) -> Result<(), String> { 258 304 let url = format!("{}/xrpc/com.atproto.repo.deleteRecord", self.base_url); 259 259 - let token = self 260 260 - .auth_token 261 261 - .as_ref() 262 262 - .ok_or("not authenticated")?; 305 305 + let token = self.auth_token.as_ref().ok_or("not authenticated")?; 263 306 264 307 let did = ""; // needs to be passed 265 308 let body = serde_json::json!({ ··· 294 337 rkey: &str, 295 338 ) -> Result<(), String> { 296 339 let url = format!("{}/xrpc/com.atproto.repo.deleteRecord", self.base_url); 297 297 - let token = self 298 298 - .auth_token 299 299 - .as_ref() 300 300 - .ok_or("not authenticated")?; 340 340 + let token = self.auth_token.as_ref().ok_or("not authenticated")?; 301 341 302 342 let body = serde_json::json!({ 303 343 "repo": did,

+430 -83

src/save.rs

reviewed

··· 3 3 use std::collections::HashMap; 4 4 use std::path::Path; 5 5 6 6 + use crate::pack::{self, PackDataType}; 6 7 use crate::pds_client::PdsClient; 7 7 - use crate::types::{FileEntry, FileKind, SaveResult, SiteRecord, COLLECTION, MANIFEST_KEY}; 8 8 + use crate::types::{ 9 9 + BlobRef, FileEntry, FileKind, PackRef, SaveResult, SiteRecord, COLLECTION, MANIFEST_KEY, 10 10 + }; 8 11 use crate::yrs_pds; 9 12 10 13 /// Compaction threshold: create new snapshot when updates exceed this count. 11 14 const COMPACTION_THRESHOLD: u32 = 50; 12 15 16 16 + /// Pending blob to be packed into a single upload. 17 17 + struct PendingBlob { 18 18 + path: String, 19 19 + data: Vec<u8>, 20 20 + data_type: PackDataType, 21 21 + } 22 22 + 13 23 /// Save a directory to PDS. 14 24 /// 15 25 /// Maintains a CRDT manifest (Yrs Map) tracking all files. Supports both 16 26 /// text files (Yrs CRDT merge) and binary files (raw blob storage). 27 27 + /// All blobs are bundled into a single pack blob upload. 17 28 pub async fn save( 18 29 dir: &Path, 19 30 client: &PdsClient, ··· 21 32 rkey: &str, 22 33 verbose: bool, 23 34 ) -> Result<SaveResult, String> { 24 24 - // Collect all files from directory (text + binary) 25 25 - let local_files = collect_files(dir)?; 35 35 + save_filtered(dir, client, did, rkey, None, None, verbose).await 36 36 + } 37 37 + 38 38 + /// Save a directory to PDS with optional include/exclude glob filters. 39 39 + pub async fn save_filtered( 40 40 + dir: &Path, 41 41 + client: &PdsClient, 42 42 + did: &str, 43 43 + rkey: &str, 44 44 + include: Option<&[String]>, 45 45 + exclude: Option<&[String]>, 46 46 + verbose: bool, 47 47 + ) -> Result<SaveResult, String> { 48 48 + // Collect files from directory with optional filters 49 49 + let local_files = collect_files_filtered(dir, include, exclude)?; 26 50 if local_files.is_empty() { 27 51 return Err("no files found".to_string()); 28 52 } 29 53 30 54 // Fetch existing record if present 31 55 let existing = client.get_record(did, COLLECTION, rkey).await?; 32 32 - let existing_site: Option<SiteRecord> = existing.as_ref().and_then(|r| { 33 33 - serde_json::from_value(r.value.clone()).ok() 34 34 - }); 56 56 + let existing_site: Option<SiteRecord> = existing 57 57 + .as_ref() 58 58 + .and_then(|r| serde_json::from_value(r.value.clone()).ok()); 35 59 let swap_cid = existing.as_ref().and_then(|r| r.cid.clone()); 36 60 37 61 // Reconstruct or create manifest 38 62 let manifest_doc = if let Some(ref site) = existing_site { 39 63 if let Some(manifest_entry) = site.files.get(MANIFEST_KEY) { 40 40 - // Load existing manifest from PDS 41 64 yrs_pds::file_entry_to_doc(manifest_entry, client, did).await? 42 65 } else { 43 43 - // Legacy record without manifest — create one from existing files 44 66 let doc = yrs_pds::new_manifest_doc(); 45 67 for (path, entry) in &site.files { 46 68 if path != MANIFEST_KEY { ··· 54 76 }; 55 77 56 78 let mut file_entries: HashMap<String, FileEntry> = HashMap::new(); 79 79 + let mut pending_blobs: Vec<PendingBlob> = Vec::new(); 57 80 let mut files_uploaded = 0; 58 81 let mut files_skipped = 0; 59 59 - let mut total_bytes: u64 = 0; 60 82 61 83 // Track which local files exist (for deletion detection) 62 84 let local_paths: std::collections::HashSet<String> = 63 85 local_files.iter().map(|(p, _, _)| p.clone()).collect(); 64 86 65 65 - // Process each local file 87 87 + // First pass: determine what needs uploading, collect blob data 66 88 for (rel_path, file_data, kind) in &local_files { 67 89 match kind { 68 90 FileKind::Text => { ··· 70 92 Ok(s) => s, 71 93 Err(_) => { 72 94 // Not valid UTF-8 — treat as binary 73 73 - let entry = save_binary_file(rel_path, file_data, client).await?; 74 74 - total_bytes += entry.snapshot_blob.size; 95 95 + let hash = hex_hash(file_data); 75 96 yrs_pds::manifest_insert(&manifest_doc, rel_path, &FileKind::Binary); 76 76 - file_entries.insert(rel_path.clone(), entry); 97 97 + pending_blobs.push(PendingBlob { 98 98 + path: rel_path.clone(), 99 99 + data: file_data.clone(), 100 100 + data_type: PackDataType::Binary, 101 101 + }); 102 102 + // Placeholder entry — will be updated with pack_ref after upload 103 103 + file_entries.insert(rel_path.clone(), placeholder_binary_entry(&hash)); 77 104 files_uploaded += 1; 78 105 continue; 79 106 } ··· 83 110 if let Some(ref site) = existing_site { 84 111 if let Some(existing_entry) = site.files.get(rel_path) { 85 112 if existing_entry.content == content { 86 86 - // Unchanged — reuse existing entry 87 113 file_entries.insert(rel_path.clone(), existing_entry.clone()); 88 114 files_skipped += 1; 89 115 if verbose { ··· 92 118 continue; 93 119 } 94 120 95 95 - // Changed — re-assert in manifest (edit wins over concurrent delete) 121 121 + // Changed — re-assert in manifest 96 122 yrs_pds::manifest_insert(&manifest_doc, rel_path, &FileKind::Text); 97 123 98 98 - // Try incremental update 124 124 + // Incremental update path 99 125 if existing_entry.updates_count < COMPACTION_THRESHOLD { 100 100 - if let Ok(entry) = incremental_update( 101 101 - existing_entry, content, client, did, verbose, 102 102 - ).await { 103 103 - total_bytes += entry.snapshot_blob.size; 104 104 - file_entries.insert(rel_path.clone(), entry); 126 126 + if let Ok(doc) = 127 127 + reconstruct_and_diff(existing_entry, content, client, did).await 128 128 + { 129 129 + let snapshot = yrs_pds::encode_snapshot(&doc); 130 130 + let sv = yrs_pds::encode_state_vector(&doc); 131 131 + let materialized = yrs_pds::materialize(&doc); 132 132 + pending_blobs.push(PendingBlob { 133 133 + path: rel_path.clone(), 134 134 + data: snapshot, 135 135 + data_type: PackDataType::Snapshot, 136 136 + }); 137 137 + file_entries.insert( 138 138 + rel_path.clone(), 139 139 + FileEntry { 140 140 + content: materialized, 141 141 + snapshot_blob: placeholder_blob_ref(), 142 142 + state_vector: yrs_pds::base64_encode(&sv), 143 143 + updates_blob: None, 144 144 + updates_count: existing_entry.updates_count + 1, 145 145 + snapshot_at: chrono::Utc::now().to_rfc3339(), 146 146 + kind: FileKind::Text, 147 147 + pack_ref: None, 148 148 + conflict_source: None, 149 149 + }, 150 150 + ); 105 151 files_uploaded += 1; 106 152 if verbose { 107 153 eprintln!("pds-yrs: incremental update {}", rel_path); ··· 114 160 eprintln!("pds-yrs: full snapshot (compaction) {}", rel_path); 115 161 } 116 162 } else { 117 117 - // New file — insert into manifest 118 163 yrs_pds::manifest_insert(&manifest_doc, rel_path, &FileKind::Text); 119 164 } 120 165 } else { 121 121 - // First save — insert into manifest 122 166 yrs_pds::manifest_insert(&manifest_doc, rel_path, &FileKind::Text); 123 167 } 124 168 125 169 // Full snapshot 126 170 let doc = yrs_pds::doc_from_text(content); 127 127 - let entry = yrs_pds::doc_to_file_entry(&doc, client, did).await?; 128 128 - total_bytes += entry.snapshot_blob.size; 129 129 - file_entries.insert(rel_path.clone(), entry); 171 171 + let snapshot = yrs_pds::encode_snapshot(&doc); 172 172 + let sv = yrs_pds::encode_state_vector(&doc); 173 173 + pending_blobs.push(PendingBlob { 174 174 + path: rel_path.clone(), 175 175 + data: snapshot, 176 176 + data_type: PackDataType::Snapshot, 177 177 + }); 178 178 + file_entries.insert( 179 179 + rel_path.clone(), 180 180 + FileEntry { 181 181 + content: content.to_string(), 182 182 + snapshot_blob: placeholder_blob_ref(), 183 183 + state_vector: yrs_pds::base64_encode(&sv), 184 184 + updates_blob: None, 185 185 + updates_count: 0, 186 186 + snapshot_at: chrono::Utc::now().to_rfc3339(), 187 187 + kind: FileKind::Text, 188 188 + pack_ref: None, 189 189 + conflict_source: None, 190 190 + }, 191 191 + ); 130 192 files_uploaded += 1; 131 193 if verbose { 132 194 eprintln!("pds-yrs: upload {}", rel_path); 133 195 } 134 196 } 135 197 FileKind::Binary => { 136 136 - // Check if binary file changed 137 198 if let Some(ref site) = existing_site { 138 199 if let Some(existing_entry) = site.files.get(rel_path) { 139 200 if existing_entry.kind == FileKind::Binary { 140 140 - // Compare by content hash (stored in content field) 141 201 let hash = hex_hash(file_data); 142 202 if existing_entry.content == hash { 143 203 file_entries.insert(rel_path.clone(), existing_entry.clone()); ··· 148 208 continue; 149 209 } 150 210 } 151 151 - // Changed binary — re-assert in manifest 152 211 yrs_pds::manifest_insert(&manifest_doc, rel_path, &FileKind::Binary); 153 212 } else { 154 213 yrs_pds::manifest_insert(&manifest_doc, rel_path, &FileKind::Binary); ··· 157 216 yrs_pds::manifest_insert(&manifest_doc, rel_path, &FileKind::Binary); 158 217 } 159 218 160 160 - let entry = save_binary_file(rel_path, file_data, client).await?; 161 161 - total_bytes += entry.snapshot_blob.size; 162 162 - file_entries.insert(rel_path.clone(), entry); 219 219 + let hash = hex_hash(file_data); 220 220 + pending_blobs.push(PendingBlob { 221 221 + path: rel_path.clone(), 222 222 + data: file_data.clone(), 223 223 + data_type: PackDataType::Binary, 224 224 + }); 225 225 + file_entries.insert(rel_path.clone(), placeholder_binary_entry(&hash)); 163 226 files_uploaded += 1; 164 227 if verbose { 165 228 eprintln!("pds-yrs: upload (binary) {}", rel_path); ··· 179 242 } 180 243 } 181 244 182 182 - // Save manifest as a FileEntry 183 183 - let manifest_entry = yrs_pds::doc_to_file_entry(&manifest_doc, client, did).await?; 184 184 - file_entries.insert(MANIFEST_KEY.to_string(), manifest_entry); 245 245 + // Add manifest snapshot to pending blobs 246 246 + let manifest_snapshot = yrs_pds::encode_snapshot(&manifest_doc); 247 247 + let manifest_sv = yrs_pds::encode_state_vector(&manifest_doc); 248 248 + pending_blobs.push(PendingBlob { 249 249 + path: MANIFEST_KEY.to_string(), 250 250 + data: manifest_snapshot, 251 251 + data_type: PackDataType::Snapshot, 252 252 + }); 253 253 + 254 254 + // Upload all blobs as a single pack (or individually if nothing to pack) 255 255 + let total_bytes; 256 256 + if pending_blobs.is_empty() { 257 257 + total_bytes = 0; 258 258 + // Still need manifest entry 259 259 + let manifest_entry = yrs_pds::doc_to_file_entry(&manifest_doc, client, did).await?; 260 260 + file_entries.insert(MANIFEST_KEY.to_string(), manifest_entry); 261 261 + } else { 262 262 + // Build pack blob 263 263 + let items: Vec<(&str, &[u8], PackDataType)> = pending_blobs 264 264 + .iter() 265 265 + .map(|pb| (pb.path.as_str(), pb.data.as_slice(), pb.data_type.clone())) 266 266 + .collect(); 267 267 + let pack_blob = pack::create_compressed_pack(&items); 268 268 + let is_compressed = pack::is_gzip(&pack_blob.data); 269 269 + total_bytes = pack_blob.data.len() as u64; 270 270 + 271 271 + // Upload pack blob — chunk if larger than AT Protocol limit 272 272 + let (blob_ref, chunk_refs) = if pack_blob.data.len() > pack::CHUNK_SIZE { 273 273 + let chunks = pack::chunk_data(&pack_blob.data); 274 274 + let mut refs = Vec::new(); 275 275 + for (i, chunk) in chunks.iter().enumerate() { 276 276 + let r = client.upload_blob(chunk.clone()).await?; 277 277 + if verbose { 278 278 + eprintln!( 279 279 + "pds-yrs: uploaded chunk {}/{} ({} bytes)", 280 280 + i + 1, 281 281 + chunks.len(), 282 282 + chunk.len() 283 283 + ); 284 284 + } 285 285 + refs.push(r); 286 286 + } 287 287 + // Use first chunk's ref as the primary blob ref 288 288 + let primary = refs[0].clone(); 289 289 + (primary, Some(refs)) 290 290 + } else { 291 291 + let r = client.upload_blob(pack_blob.data).await?; 292 292 + (r, None) 293 293 + }; 294 294 + 295 295 + if verbose { 296 296 + eprintln!( 297 297 + "pds-yrs: uploaded pack blob ({} bytes, {} entries{})", 298 298 + total_bytes, 299 299 + pack_blob.entries.len(), 300 300 + if chunk_refs.is_some() { 301 301 + ", chunked" 302 302 + } else { 303 303 + "" 304 304 + } 305 305 + ); 306 306 + } 307 307 + 308 308 + // Update file entries with pack refs 309 309 + for entry in &pack_blob.entries { 310 310 + let pack_ref = PackRef { 311 311 + blob: blob_ref.clone(), 312 312 + offset: entry.offset, 313 313 + length: entry.length, 314 314 + compressed: is_compressed, 315 315 + chunks: chunk_refs.clone(), 316 316 + }; 317 317 + 318 318 + if entry.path == MANIFEST_KEY { 319 319 + // Manifest entry 320 320 + let manifest_content = yrs_pds::materialize_manifest_content(&manifest_doc); 321 321 + file_entries.insert( 322 322 + MANIFEST_KEY.to_string(), 323 323 + FileEntry { 324 324 + content: manifest_content, 325 325 + snapshot_blob: blob_ref.clone(), 326 326 + state_vector: yrs_pds::base64_encode(&manifest_sv), 327 327 + updates_blob: None, 328 328 + updates_count: 0, 329 329 + snapshot_at: chrono::Utc::now().to_rfc3339(), 330 330 + kind: FileKind::Text, 331 331 + pack_ref: Some(pack_ref), 332 332 + conflict_source: None, 333 333 + }, 334 334 + ); 335 335 + } else if let Some(fe) = file_entries.get_mut(&entry.path) { 336 336 + // Update placeholder blob ref with actual pack ref 337 337 + fe.snapshot_blob = blob_ref.clone(); 338 338 + fe.pack_ref = Some(pack_ref); 339 339 + } 340 340 + } 341 341 + } 185 342 186 343 // Build SiteRecord 187 344 let now = chrono::Utc::now().to_rfc3339(); ··· 191 348 updated_at: now, 192 349 }; 193 350 194 194 - let record_json = serde_json::to_value(&record) 195 195 - .map_err(|e| format!("serialize SiteRecord: {}", e))?; 351 351 + let record_json = 352 352 + serde_json::to_value(&record).map_err(|e| format!("serialize SiteRecord: {}", e))?; 196 353 197 354 client 198 355 .put_record(did, COLLECTION, rkey, record_json, swap_cid) ··· 205 362 }) 206 363 } 207 364 208 208 - /// Save a binary file as a raw blob upload. 209 209 - async fn save_binary_file( 210 210 - _rel_path: &str, 211 211 - data: &[u8], 212 212 - client: &PdsClient, 213 213 - ) -> Result<FileEntry, String> { 214 214 - let blob_ref = client.upload_blob(data.to_vec()).await?; 215 215 - let hash = hex_hash(data); 216 216 - let now = chrono::Utc::now().to_rfc3339(); 365 365 + /// Placeholder BlobRef — will be replaced with pack ref after upload. 366 366 + fn placeholder_blob_ref() -> BlobRef { 367 367 + BlobRef::new( 368 368 + "pending".to_string(), 369 369 + "application/octet-stream".to_string(), 370 370 + 0, 371 371 + ) 372 372 + } 217 373 218 218 - Ok(FileEntry { 219 219 - content: hash, 220 220 - snapshot_blob: blob_ref, 374 374 + /// Placeholder binary FileEntry. 375 375 + fn placeholder_binary_entry(hash: &str) -> FileEntry { 376 376 + FileEntry { 377 377 + content: hash.to_string(), 378 378 + snapshot_blob: placeholder_blob_ref(), 221 379 state_vector: String::new(), 222 380 updates_blob: None, 223 381 updates_count: 0, 224 224 - snapshot_at: now, 382 382 + snapshot_at: chrono::Utc::now().to_rfc3339(), 225 383 kind: FileKind::Binary, 226 384 pack_ref: None, 227 385 conflict_source: None, 228 228 - }) 386 386 + } 229 387 } 230 388 231 231 - /// Attempt an incremental update to an existing FileEntry. 232 232 - async fn incremental_update( 389 389 + /// Reconstruct a doc from existing entry, apply diff, return updated doc. 390 390 + async fn reconstruct_and_diff( 233 391 existing: &FileEntry, 234 392 new_content: &str, 235 393 client: &PdsClient, 236 394 did: &str, 237 237 - _verbose: bool, 238 238 - ) -> Result<FileEntry, String> { 395 395 + ) -> Result<yrs::Doc, String> { 239 396 let doc = yrs_pds::file_entry_to_doc(existing, client, did).await?; 240 397 let old_content = yrs_pds::materialize(&doc); 241 398 apply_text_diff(&doc, &old_content, new_content); 242 242 - 243 243 - let content = yrs_pds::materialize(&doc); 244 244 - let snapshot = yrs_pds::encode_snapshot(&doc); 245 245 - let sv = yrs_pds::encode_state_vector(&doc); 246 246 - let snapshot_blob = client.upload_blob(snapshot).await?; 247 247 - 248 248 - let now = chrono::Utc::now().to_rfc3339(); 249 249 - Ok(FileEntry { 250 250 - content, 251 251 - snapshot_blob, 252 252 - state_vector: yrs_pds::base64_encode(&sv), 253 253 - updates_blob: None, 254 254 - updates_count: existing.updates_count + 1, 255 255 - snapshot_at: now, 256 256 - kind: FileKind::Text, 257 257 - pack_ref: None, 258 258 - conflict_source: None, 259 259 - }) 399 399 + Ok(doc) 260 400 } 261 401 262 402 /// Apply a text diff to a Yrs Doc as character-level operations. ··· 292 432 293 433 /// Collect all files from a directory, returning (relative_path, data, kind). 294 434 pub fn collect_files(dir: &Path) -> Result<Vec<(String, Vec<u8>, FileKind)>, String> { 435 435 + collect_files_filtered(dir, None, None) 436 436 + } 437 437 + 438 438 + /// Collect files with optional include/exclude glob filters. 439 439 + pub fn collect_files_filtered( 440 440 + dir: &Path, 441 441 + include: Option<&[String]>, 442 442 + exclude: Option<&[String]>, 443 443 + ) -> Result<Vec<(String, Vec<u8>, FileKind)>, String> { 295 444 let mut files = Vec::new(); 296 445 collect_recursive(dir, dir, &mut files)?; 297 446 files.sort_by(|a, b| a.0.cmp(&b.0)); 447 447 + 448 448 + // Apply filters 449 449 + if include.is_some() || exclude.is_some() { 450 450 + files.retain(|(path, _, _)| { 451 451 + if let Some(includes) = include { 452 452 + if !includes.iter().any(|pattern| glob_match(pattern, path)) { 453 453 + return false; 454 454 + } 455 455 + } 456 456 + if let Some(excludes) = exclude { 457 457 + if excludes.iter().any(|pattern| glob_match(pattern, path)) { 458 458 + return false; 459 459 + } 460 460 + } 461 461 + true 462 462 + }); 463 463 + } 464 464 + 298 465 Ok(files) 299 466 } 300 467 468 468 + /// Simple glob matching (supports * and **). 469 469 + fn glob_match(pattern: &str, path: &str) -> bool { 470 470 + if pattern == "*" { 471 471 + return true; 472 472 + } 473 473 + if pattern.starts_with("*.") { 474 474 + // Extension match: *.md matches any .md file 475 475 + let ext = &pattern[1..]; // ".md" 476 476 + return path.ends_with(ext); 477 477 + } 478 478 + if pattern.contains("**") { 479 479 + // ** matches any path segment 480 480 + let parts: Vec<&str> = pattern.split("**").collect(); 481 481 + if parts.len() == 2 { 482 482 + let prefix = parts[0].trim_end_matches('/'); 483 483 + let suffix = parts[1].trim_start_matches('/'); 484 484 + let prefix_ok = prefix.is_empty() || path.starts_with(prefix); 485 485 + let suffix_ok = suffix.is_empty() || path.ends_with(suffix); 486 486 + return prefix_ok && suffix_ok; 487 487 + } 488 488 + } 489 489 + // Exact match 490 490 + pattern == path 491 491 + } 492 492 + 301 493 fn collect_recursive( 302 494 base: &Path, 303 495 current: &Path, 304 496 files: &mut Vec<(String, Vec<u8>, FileKind)>, 305 497 ) -> Result<(), String> { 306 306 - let entries = std::fs::read_dir(current) 307 307 - .map_err(|e| format!("read dir {:?}: {}", current, e))?; 498 498 + let entries = 499 499 + std::fs::read_dir(current).map_err(|e| format!("read dir {:?}: {}", current, e))?; 308 500 309 501 for entry in entries { 310 502 let entry = entry.map_err(|e| format!("read dir entry: {}", e))?; ··· 312 504 313 505 if path.is_dir() { 314 506 let name = path.file_name().unwrap().to_str().unwrap_or(""); 315 315 - // Skip hidden directories, common non-content dirs, and .pds-yrs state 316 507 if name.starts_with('.') || name == "node_modules" || name == "target" { 317 508 continue; 318 509 } ··· 326 517 .to_str() 327 518 .ok_or("non-utf8 path")? 328 519 .to_string(); 329 329 - let data = std::fs::read(&path) 330 330 - .map_err(|e| format!("read {:?}: {}", path, e))?; 520 520 + let data = std::fs::read(&path).map_err(|e| format!("read {:?}: {}", path, e))?; 331 521 files.push((rel_path, data, kind)); 332 522 } 333 523 } ··· 335 525 } 336 526 337 527 /// Simple hex hash of data (for binary file change detection). 338 338 - fn hex_hash(data: &[u8]) -> String { 339 339 - // Simple FNV-1a hash — not cryptographic, just for change detection 528 528 + pub fn hex_hash(data: &[u8]) -> String { 340 529 let mut hash: u64 = 0xcbf29ce484222325; 341 530 for &byte in data { 342 531 hash ^= byte as u64; ··· 359 548 360 549 let files = collect_files(tmp.path()).unwrap(); 361 550 let paths: Vec<(&str, &FileKind)> = files.iter().map(|(p, _, k)| (p.as_str(), k)).collect(); 362 362 - assert!(paths.iter().any(|(p, k)| *p == "index.md" && **k == FileKind::Text)); 363 363 - assert!(paths.iter().any(|(p, k)| *p == "blog/post.md" && **k == FileKind::Text)); 364 364 - assert!(paths.iter().any(|(p, k)| *p == "image.png" && **k == FileKind::Binary)); 551 551 + assert!(paths 552 552 + .iter() 553 553 + .any(|(p, k)| *p == "index.md" && **k == FileKind::Text)); 554 554 + assert!(paths 555 555 + .iter() 556 556 + .any(|(p, k)| *p == "blog/post.md" && **k == FileKind::Text)); 557 557 + assert!(paths 558 558 + .iter() 559 559 + .any(|(p, k)| *p == "image.png" && **k == FileKind::Binary)); 365 560 } 366 561 367 562 #[test] ··· 392 587 let h3 = hex_hash(b"world"); 393 588 assert_eq!(h1, h2); 394 589 assert_ne!(h1, h3); 590 590 + } 591 591 + 592 592 + #[test] 593 593 + fn glob_match_extension() { 594 594 + assert!(glob_match("*.md", "docs/index.md")); 595 595 + assert!(glob_match("*.md", "index.md")); 596 596 + assert!(!glob_match("*.md", "image.png")); 597 597 + } 598 598 + 599 599 + #[test] 600 600 + fn glob_match_doublestar() { 601 601 + assert!(glob_match("docs/**", "docs/index.md")); 602 602 + assert!(glob_match("docs/**", "docs/sub/file.md")); 603 603 + assert!(!glob_match("docs/**", "src/main.rs")); 604 604 + } 605 605 + 606 606 + #[test] 607 607 + fn glob_match_exact() { 608 608 + assert!(glob_match("README.md", "README.md")); 609 609 + assert!(!glob_match("README.md", "docs/README.md")); 610 610 + } 611 611 + 612 612 + #[test] 613 613 + fn collect_files_filtered_include() { 614 614 + let tmp = tempfile::tempdir().unwrap(); 615 615 + std::fs::write(tmp.path().join("index.md"), "# Home").unwrap(); 616 616 + std::fs::write(tmp.path().join("image.png"), "binary").unwrap(); 617 617 + std::fs::write(tmp.path().join("style.css"), "body{}").unwrap(); 618 618 + 619 619 + let include = vec!["*.md".to_string()]; 620 620 + let files = collect_files_filtered(tmp.path(), Some(&include), None).unwrap(); 621 621 + assert_eq!(files.len(), 1); 622 622 + assert_eq!(files[0].0, "index.md"); 623 623 + } 624 624 + 625 625 + #[test] 626 626 + fn collect_files_filtered_exclude() { 627 627 + let tmp = tempfile::tempdir().unwrap(); 628 628 + std::fs::write(tmp.path().join("index.md"), "# Home").unwrap(); 629 629 + std::fs::write(tmp.path().join("image.png"), "binary").unwrap(); 630 630 + 631 631 + let exclude = vec!["*.png".to_string()]; 632 632 + let files = collect_files_filtered(tmp.path(), None, Some(&exclude)).unwrap(); 633 633 + assert_eq!(files.len(), 1); 634 634 + assert_eq!(files[0].0, "index.md"); 635 635 + } 636 636 + 637 637 + #[test] 638 638 + fn binary_file_hash_change_detection() { 639 639 + // Same content → same hash 640 640 + let h1 = hex_hash(b"\x89PNG\r\n\x1a\nsome image data"); 641 641 + let h2 = hex_hash(b"\x89PNG\r\n\x1a\nsome image data"); 642 642 + assert_eq!(h1, h2); 643 643 + 644 644 + // Modified content → different hash 645 645 + let h3 = hex_hash(b"\x89PNG\r\n\x1a\nmodified image data"); 646 646 + assert_ne!(h1, h3); 647 647 + } 648 648 + 649 649 + #[test] 650 650 + fn placeholder_binary_entry_has_correct_kind() { 651 651 + let entry = placeholder_binary_entry("abc123"); 652 652 + assert_eq!(entry.kind, FileKind::Binary); 653 653 + assert_eq!(entry.content, "abc123"); 654 654 + assert!(entry.state_vector.is_empty()); 655 655 + assert!(entry.updates_blob.is_none()); 656 656 + } 657 657 + 658 658 + #[test] 659 659 + fn collect_mixed_text_and_binary() { 660 660 + let tmp = tempfile::tempdir().unwrap(); 661 661 + std::fs::write(tmp.path().join("readme.md"), "# Readme").unwrap(); 662 662 + std::fs::write(tmp.path().join("logo.png"), &[0x89, 0x50, 0x4E, 0x47]).unwrap(); 663 663 + std::fs::write(tmp.path().join("app.js"), "console.log('hi')").unwrap(); 664 664 + std::fs::write(tmp.path().join("photo.jpg"), &[0xFF, 0xD8, 0xFF]).unwrap(); 665 665 + std::fs::write(tmp.path().join("data.json"), r#"{"key":"value"}"#).unwrap(); 666 666 + 667 667 + let files = collect_files(tmp.path()).unwrap(); 668 668 + assert_eq!(files.len(), 5); 669 669 + 670 670 + let text_count = files 671 671 + .iter() 672 672 + .filter(|(_, _, k)| *k == FileKind::Text) 673 673 + .count(); 674 674 + let binary_count = files 675 675 + .iter() 676 676 + .filter(|(_, _, k)| *k == FileKind::Binary) 677 677 + .count(); 678 678 + assert_eq!(text_count, 3); // md, js, json 679 679 + assert_eq!(binary_count, 2); // png, jpg 680 680 + } 681 681 + 682 682 + #[test] 683 683 + fn binary_hash_for_empty_data() { 684 684 + let h1 = hex_hash(b""); 685 685 + let h2 = hex_hash(b""); 686 686 + assert_eq!(h1, h2); 687 687 + // Should be a valid 16-char hex string 688 688 + assert_eq!(h1.len(), 16); 689 689 + } 690 690 + 691 691 + #[test] 692 692 + fn binary_file_collect_preserves_raw_bytes() { 693 693 + // binary files should be collected as raw bytes, not interpreted 694 694 + let tmp = tempfile::tempdir().unwrap(); 695 695 + let png_bytes: Vec<u8> = vec![0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, 0x00, 0xFF]; 696 696 + std::fs::write(tmp.path().join("image.png"), &png_bytes).unwrap(); 697 697 + 698 698 + let files = collect_files(tmp.path()).unwrap(); 699 699 + assert_eq!(files.len(), 1); 700 700 + assert_eq!(files[0].0, "image.png"); 701 701 + assert_eq!(files[0].1, png_bytes); 702 702 + assert_eq!(files[0].2, FileKind::Binary); 703 703 + } 704 704 + 705 705 + #[test] 706 706 + fn binary_hash_changes_on_single_byte_diff() { 707 707 + let data_a = vec![0u8; 1024]; 708 708 + let mut data_b = data_a.clone(); 709 709 + data_b[512] = 1; 710 710 + 711 711 + assert_ne!(hex_hash(&data_a), hex_hash(&data_b)); 712 712 + } 713 713 + 714 714 + #[test] 715 715 + fn placeholder_binary_entry_fields() { 716 716 + let entry = placeholder_binary_entry("deadbeef"); 717 717 + assert_eq!(entry.kind, FileKind::Binary); 718 718 + assert_eq!(entry.content, "deadbeef"); 719 719 + assert!(entry.pack_ref.is_none()); 720 720 + assert!(entry.conflict_source.is_none()); 721 721 + assert_eq!(entry.updates_count, 0); 722 722 + } 723 723 + 724 724 + #[test] 725 725 + fn collect_files_filtered_include_and_exclude_combined() { 726 726 + let tmp = tempfile::tempdir().unwrap(); 727 727 + std::fs::write(tmp.path().join("readme.md"), "# Readme").unwrap(); 728 728 + std::fs::write(tmp.path().join("draft.md"), "draft").unwrap(); 729 729 + std::fs::write(tmp.path().join("image.png"), "binary").unwrap(); 730 730 + std::fs::write(tmp.path().join("notes.txt"), "notes").unwrap(); 731 731 + 732 732 + // Include all text-like, exclude drafts 733 733 + let include = vec!["*.md".to_string(), "*.txt".to_string()]; 734 734 + let exclude = vec!["draft.md".to_string()]; 735 735 + let files = collect_files_filtered(tmp.path(), Some(&include), Some(&exclude)).unwrap(); 736 736 + 737 737 + let paths: Vec<&str> = files.iter().map(|(p, _, _)| p.as_str()).collect(); 738 738 + assert!(paths.contains(&"readme.md")); 739 739 + assert!(paths.contains(&"notes.txt")); 740 740 + assert!(!paths.contains(&"draft.md")); 741 741 + assert!(!paths.contains(&"image.png")); 395 742 } 396 743 }

+308

src/sync.rs

reviewed

··· 1 1 + //! Real-time sync relay: PDS as WebRTC fallback. 2 2 + //! 3 3 + //! When two live peers can't form a WebRTC connection, the PDS acts as a relay. 4 4 + //! Each peer polls in a loop: load remote changes → apply local edits → save diffs. 5 5 + //! 6 6 + //! Two modes: 7 7 + //! - **Normal mode**: Full materialized save/load on each cycle (default). 8 8 + //! - **Streaming mode**: Only upload Yrs diffs, skip materialization. 9 9 + //! Periodic materialization every N cycles or on explicit request. 10 10 + 11 11 + use std::path::Path; 12 12 + use std::time::Duration; 13 13 + 14 14 + use crate::load; 15 15 + use crate::pds_client::PdsClient; 16 16 + use crate::save; 17 17 + 18 18 + /// Sync configuration. 19 19 + pub struct SyncConfig { 20 20 + /// Directory to sync. 21 21 + pub dir: String, 22 22 + /// Poll interval between sync cycles. 23 23 + pub interval: Duration, 24 24 + /// Maximum number of cycles (None = infinite). 25 25 + pub max_cycles: Option<u32>, 26 26 + /// Materialize every N cycles in streaming mode (0 = every cycle). 27 27 + pub materialize_interval: u32, 28 28 + /// Include glob patterns for file filtering. 29 29 + pub include: Vec<String>, 30 30 + /// Exclude glob patterns for file filtering. 31 31 + pub exclude: Vec<String>, 32 32 + /// Show progress. 33 33 + pub verbose: bool, 34 34 + } 35 35 + 36 36 + impl Default for SyncConfig { 37 37 + fn default() -> Self { 38 38 + Self { 39 39 + dir: ".".to_string(), 40 40 + interval: Duration::from_secs(3), 41 41 + max_cycles: None, 42 42 + materialize_interval: 0, 43 43 + include: Vec::new(), 44 44 + exclude: Vec::new(), 45 45 + verbose: false, 46 46 + } 47 47 + } 48 48 + } 49 49 + 50 50 + /// Result of a single sync cycle. 51 51 + #[derive(Debug)] 52 52 + pub struct SyncCycleResult { 53 53 + /// Files loaded from remote. 54 54 + pub files_loaded: usize, 55 55 + /// Files uploaded to remote. 56 56 + pub files_uploaded: usize, 57 57 + /// Whether this cycle performed materialization. 58 58 + pub materialized: bool, 59 59 + } 60 60 + 61 61 + /// Run the sync loop. 62 62 + /// 63 63 + /// Each cycle: load remote changes, then save local changes. 64 64 + /// The PDS acts as a relay between peers. 65 65 + pub async fn sync_loop( 66 66 + client: &PdsClient, 67 67 + did: &str, 68 68 + rkey: &str, 69 69 + config: &SyncConfig, 70 70 + ) -> Result<Vec<SyncCycleResult>, String> { 71 71 + let dir = Path::new(&config.dir); 72 72 + let mut results = Vec::new(); 73 73 + let mut cycle = 0u32; 74 74 + 75 75 + loop { 76 76 + if let Some(max) = config.max_cycles { 77 77 + if cycle >= max { 78 78 + break; 79 79 + } 80 80 + } 81 81 + 82 82 + let result = sync_cycle(client, did, rkey, dir, cycle, config).await?; 83 83 + 84 84 + if config.verbose { 85 85 + eprintln!( 86 86 + "pds-yrs sync: cycle {} — loaded {}, uploaded {}{}", 87 87 + cycle, 88 88 + result.files_loaded, 89 89 + result.files_uploaded, 90 90 + if result.materialized { 91 91 + " (materialized)" 92 92 + } else { 93 93 + "" 94 94 + } 95 95 + ); 96 96 + } 97 97 + 98 98 + results.push(result); 99 99 + cycle += 1; 100 100 + 101 101 + if config.max_cycles.is_some() && cycle >= config.max_cycles.unwrap() { 102 102 + break; 103 103 + } 104 104 + 105 105 + tokio::time::sleep(config.interval).await; 106 106 + } 107 107 + 108 108 + Ok(results) 109 109 + } 110 110 + 111 111 + /// Execute a single sync cycle: load then save. 112 112 + async fn sync_cycle( 113 113 + client: &PdsClient, 114 114 + did: &str, 115 115 + rkey: &str, 116 116 + dir: &Path, 117 117 + cycle: u32, 118 118 + config: &SyncConfig, 119 119 + ) -> Result<SyncCycleResult, String> { 120 120 + // Ensure output directory exists 121 121 + if !dir.exists() { 122 122 + std::fs::create_dir_all(dir).map_err(|e| format!("create sync dir: {}", e))?; 123 123 + } 124 124 + 125 125 + // Load remote changes 126 126 + let load_result = load::load(client, did, rkey, dir, config.verbose).await; 127 127 + let files_loaded = match load_result { 128 128 + Ok(r) => r.files_loaded, 129 129 + Err(e) => { 130 130 + // Record not found on first cycle is OK — we'll create it 131 131 + if e.contains("not found") && cycle == 0 { 132 132 + 0 133 133 + } else { 134 134 + return Err(e); 135 135 + } 136 136 + } 137 137 + }; 138 138 + 139 139 + // Check if there are local files to save 140 140 + let inc = if config.include.is_empty() { 141 141 + None 142 142 + } else { 143 143 + Some(config.include.as_slice()) 144 144 + }; 145 145 + let exc = if config.exclude.is_empty() { 146 146 + None 147 147 + } else { 148 148 + Some(config.exclude.as_slice()) 149 149 + }; 150 150 + let local_files = save::collect_files_filtered(dir, inc, exc)?; 151 151 + let files_uploaded; 152 152 + let materialized; 153 153 + 154 154 + if local_files.is_empty() { 155 155 + files_uploaded = 0; 156 156 + materialized = false; 157 157 + } else { 158 158 + // Save local changes with filters 159 159 + let save_result = 160 160 + save::save_filtered(dir, client, did, rkey, inc, exc, config.verbose).await?; 161 161 + files_uploaded = save_result.files_uploaded; 162 162 + 163 163 + // Determine if this cycle should materialize 164 164 + materialized = config.materialize_interval == 0 || cycle % config.materialize_interval == 0; 165 165 + } 166 166 + 167 167 + Ok(SyncCycleResult { 168 168 + files_loaded, 169 169 + files_uploaded, 170 170 + materialized, 171 171 + }) 172 172 + } 173 173 + 174 174 + #[cfg(test)] 175 175 + mod tests { 176 176 + use super::*; 177 177 + 178 178 + #[test] 179 179 + fn sync_config_default() { 180 180 + let config = SyncConfig::default(); 181 181 + assert_eq!(config.interval, Duration::from_secs(3)); 182 182 + assert_eq!(config.max_cycles, None); 183 183 + assert_eq!(config.materialize_interval, 0); 184 184 + } 185 185 + 186 186 + #[test] 187 187 + fn sync_config_custom() { 188 188 + let config = SyncConfig { 189 189 + dir: "/tmp/test".to_string(), 190 190 + interval: Duration::from_millis(500), 191 191 + max_cycles: Some(10), 192 192 + materialize_interval: 5, 193 193 + include: vec!["*.md".to_string()], 194 194 + exclude: vec!["*.tmp".to_string()], 195 195 + verbose: true, 196 196 + }; 197 197 + assert_eq!(config.max_cycles, Some(10)); 198 198 + assert_eq!(config.materialize_interval, 5); 199 199 + assert_eq!(config.include, vec!["*.md"]); 200 200 + assert_eq!(config.exclude, vec!["*.tmp"]); 201 201 + } 202 202 + 203 203 + #[test] 204 204 + fn sync_config_default_has_empty_filters() { 205 205 + let config = SyncConfig::default(); 206 206 + assert!(config.include.is_empty()); 207 207 + assert!(config.exclude.is_empty()); 208 208 + } 209 209 + 210 210 + #[test] 211 211 + fn materialization_schedule_every_cycle() { 212 212 + // materialize_interval=0 means every cycle materializes 213 213 + let interval = 0u32; 214 214 + for cycle in 0..10 { 215 215 + let should_materialize = interval == 0 || cycle % interval == 0; 216 216 + assert!(should_materialize, "cycle {} should materialize", cycle); 217 217 + } 218 218 + } 219 219 + 220 220 + #[test] 221 221 + fn materialization_schedule_periodic() { 222 222 + // materialize_interval=3 means every 3rd cycle 223 223 + let interval = 3u32; 224 224 + let expected = vec![ 225 225 + true, false, false, true, false, false, true, false, false, true, 226 226 + ]; 227 227 + for (cycle, &expect) in expected.iter().enumerate() { 228 228 + let should = interval == 0 || (cycle as u32) % interval == 0; 229 229 + assert_eq!(should, expect, "cycle {} materialization mismatch", cycle); 230 230 + } 231 231 + } 232 232 + 233 233 + #[test] 234 234 + fn sync_cycle_result_fields() { 235 235 + let result = SyncCycleResult { 236 236 + files_loaded: 3, 237 237 + files_uploaded: 5, 238 238 + materialized: true, 239 239 + }; 240 240 + assert_eq!(result.files_loaded, 3); 241 241 + assert_eq!(result.files_uploaded, 5); 242 242 + assert!(result.materialized); 243 243 + } 244 244 + 245 245 + #[test] 246 246 + fn sync_config_filter_integration() { 247 247 + // verify that filter config translates correctly to option slices 248 248 + let config = SyncConfig { 249 249 + include: vec!["*.md".to_string(), "*.txt".to_string()], 250 250 + exclude: vec!["draft/**".to_string()], 251 251 + ..Default::default() 252 252 + }; 253 253 + 254 254 + let inc = if config.include.is_empty() { 255 255 + None 256 256 + } else { 257 257 + Some(config.include.as_slice()) 258 258 + }; 259 259 + let exc = if config.exclude.is_empty() { 260 260 + None 261 261 + } else { 262 262 + Some(config.exclude.as_slice()) 263 263 + }; 264 264 + 265 265 + assert!(inc.is_some()); 266 266 + assert_eq!(inc.unwrap().len(), 2); 267 267 + assert!(exc.is_some()); 268 268 + assert_eq!(exc.unwrap().len(), 1); 269 269 + } 270 270 + 271 271 + #[test] 272 272 + fn sync_config_empty_filter_gives_none() { 273 273 + let config = SyncConfig::default(); 274 274 + let inc = if config.include.is_empty() { 275 275 + None 276 276 + } else { 277 277 + Some(config.include.as_slice()) 278 278 + }; 279 279 + let exc = if config.exclude.is_empty() { 280 280 + None 281 281 + } else { 282 282 + Some(config.exclude.as_slice()) 283 283 + }; 284 284 + assert!(inc.is_none()); 285 285 + assert!(exc.is_none()); 286 286 + } 287 287 + 288 288 + #[test] 289 289 + fn sync_max_cycles_zero_runs_nothing() { 290 290 + // max_cycles=Some(0) should result in zero iterations 291 291 + let config = SyncConfig { 292 292 + max_cycles: Some(0), 293 293 + ..Default::default() 294 294 + }; 295 295 + let mut cycle = 0u32; 296 296 + let mut ran = false; 297 297 + loop { 298 298 + if let Some(max) = config.max_cycles { 299 299 + if cycle >= max { 300 300 + break; 301 301 + } 302 302 + } 303 303 + ran = true; 304 304 + cycle += 1; 305 305 + } 306 306 + assert!(!ran, "should not have run any cycle"); 307 307 + } 308 308 + }

+145 -4

src/types.rs

reviewed

··· 77 77 pub offset: u64, 78 78 /// Length of data within the pack blob. 79 79 pub length: u64, 80 80 + /// Whether the pack blob is gzip-compressed. 81 81 + #[serde(default, skip_serializing_if = "std::ops::Not::not")] 82 82 + pub compressed: bool, 83 83 + /// For chunked packs (>40MB), ordered list of chunk blob refs. 84 84 + /// When present, `blob` is unused — reassemble from chunks instead. 85 85 + #[serde(skip_serializing_if = "Option::is_none")] 86 86 + pub chunks: Option<Vec<BlobRef>>, 87 87 + } 88 88 + 89 89 + /// A chunked blob — multiple BlobRefs that form a single logical blob. 90 90 + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] 91 91 + pub struct ChunkedBlob { 92 92 + /// Ordered list of chunk blob refs. 93 93 + pub parts: Vec<BlobRef>, 94 94 + /// Total size of all chunks combined. 95 95 + pub total_size: u64, 80 96 } 81 97 82 98 /// AT Protocol blob reference. ··· 134 150 135 151 #[test] 136 152 fn blob_ref_serialization_round_trip() { 137 137 - let blob = BlobRef::new("bafyabc123".to_string(), "application/octet-stream".to_string(), 1024); 153 153 + let blob = BlobRef::new( 154 154 + "bafyabc123".to_string(), 155 155 + "application/octet-stream".to_string(), 156 156 + 1024, 157 157 + ); 138 158 let json = serde_json::to_string(&blob).unwrap(); 139 159 assert!(json.contains("\"$type\":\"blob\"")); 140 160 assert!(json.contains("\"$link\":\"bafyabc123\"")); ··· 146 166 fn file_entry_serialization() { 147 167 let entry = FileEntry { 148 168 content: "Hello world".to_string(), 149 149 - snapshot_blob: BlobRef::new("bafysnap".to_string(), "application/octet-stream".to_string(), 100), 169 169 + snapshot_blob: BlobRef::new( 170 170 + "bafysnap".to_string(), 171 171 + "application/octet-stream".to_string(), 172 172 + 100, 173 173 + ), 150 174 state_vector: "AQID".to_string(), 151 175 updates_blob: None, 152 176 updates_count: 0, ··· 169 193 fn binary_file_entry_serialization() { 170 194 let entry = FileEntry { 171 195 content: String::new(), 172 172 - snapshot_blob: BlobRef::new("bafybin".to_string(), "application/octet-stream".to_string(), 5000), 196 196 + snapshot_blob: BlobRef::new( 197 197 + "bafybin".to_string(), 198 198 + "application/octet-stream".to_string(), 199 199 + 5000, 200 200 + ), 173 201 state_vector: String::new(), 174 202 updates_blob: None, 175 203 updates_count: 0, ··· 191 219 "index.md".to_string(), 192 220 FileEntry { 193 221 content: "# Home".to_string(), 194 194 - snapshot_blob: BlobRef::new("bafyindex".to_string(), "application/octet-stream".to_string(), 50), 222 222 + snapshot_blob: BlobRef::new( 223 223 + "bafyindex".to_string(), 224 224 + "application/octet-stream".to_string(), 225 225 + 50, 226 226 + ), 195 227 state_vector: "AQID".to_string(), 196 228 updates_blob: None, 197 229 updates_count: 0, ··· 210 242 let deserialized: SiteRecord = serde_json::from_str(&json).unwrap(); 211 243 assert_eq!(deserialized.name, "my-site"); 212 244 assert!(deserialized.files.contains_key("index.md")); 245 245 + } 246 246 + 247 247 + #[test] 248 248 + fn pack_ref_serialization() { 249 249 + let pack_ref = PackRef { 250 250 + blob: BlobRef::new( 251 251 + "bafypack".to_string(), 252 252 + "application/octet-stream".to_string(), 253 253 + 5000, 254 254 + ), 255 255 + offset: 100, 256 256 + length: 200, 257 257 + compressed: true, 258 258 + chunks: None, 259 259 + }; 260 260 + let json = serde_json::to_string(&pack_ref).unwrap(); 261 261 + assert!(json.contains("\"compressed\":true")); 262 262 + let deserialized: PackRef = serde_json::from_str(&json).unwrap(); 263 263 + assert_eq!(deserialized.offset, 100); 264 264 + assert_eq!(deserialized.length, 200); 265 265 + assert!(deserialized.compressed); 266 266 + assert!(deserialized.chunks.is_none()); 267 267 + } 268 268 + 269 269 + #[test] 270 270 + fn pack_ref_compressed_false_omitted() { 271 271 + // compressed=false should be skipped in serialization 272 272 + let pack_ref = PackRef { 273 273 + blob: BlobRef::new( 274 274 + "bafypack".to_string(), 275 275 + "application/octet-stream".to_string(), 276 276 + 1000, 277 277 + ), 278 278 + offset: 0, 279 279 + length: 100, 280 280 + compressed: false, 281 281 + chunks: None, 282 282 + }; 283 283 + let json = serde_json::to_string(&pack_ref).unwrap(); 284 284 + assert!( 285 285 + !json.contains("compressed"), 286 286 + "compressed=false should be omitted" 287 287 + ); 288 288 + } 289 289 + 290 290 + #[test] 291 291 + fn pack_ref_with_chunks() { 292 292 + let pack_ref = PackRef { 293 293 + blob: BlobRef::new( 294 294 + "bafychunk0".to_string(), 295 295 + "application/octet-stream".to_string(), 296 296 + 40000000, 297 297 + ), 298 298 + offset: 0, 299 299 + length: 500, 300 300 + compressed: true, 301 301 + chunks: Some(vec![ 302 302 + BlobRef::new( 303 303 + "bafychunk0".to_string(), 304 304 + "application/octet-stream".to_string(), 305 305 + 40000000, 306 306 + ), 307 307 + BlobRef::new( 308 308 + "bafychunk1".to_string(), 309 309 + "application/octet-stream".to_string(), 310 310 + 10000000, 311 311 + ), 312 312 + ]), 313 313 + }; 314 314 + let json = serde_json::to_string(&pack_ref).unwrap(); 315 315 + assert!(json.contains("bafychunk1")); 316 316 + let deserialized: PackRef = serde_json::from_str(&json).unwrap(); 317 317 + assert_eq!(deserialized.chunks.as_ref().unwrap().len(), 2); 318 318 + } 319 319 + 320 320 + #[test] 321 321 + fn pack_ref_backward_compat_no_compressed_field() { 322 322 + // JSON without compressed or chunks fields should deserialize with defaults 323 323 + let json = r#"{ 324 324 + "blob": {"$type":"blob","ref":{"$link":"bafyold"},"mimeType":"application/octet-stream","size":100}, 325 325 + "offset": 0, 326 326 + "length": 50 327 327 + }"#; 328 328 + let pack_ref: PackRef = serde_json::from_str(json).unwrap(); 329 329 + assert!(!pack_ref.compressed); 330 330 + assert!(pack_ref.chunks.is_none()); 331 331 + } 332 332 + 333 333 + #[test] 334 334 + fn chunked_blob_serialization() { 335 335 + let chunked = ChunkedBlob { 336 336 + parts: vec![ 337 337 + BlobRef::new( 338 338 + "bafyp1".to_string(), 339 339 + "application/octet-stream".to_string(), 340 340 + 40000000, 341 341 + ), 342 342 + BlobRef::new( 343 343 + "bafyp2".to_string(), 344 344 + "application/octet-stream".to_string(), 345 345 + 5000000, 346 346 + ), 347 347 + ], 348 348 + total_size: 45000000, 349 349 + }; 350 350 + let json = serde_json::to_string(&chunked).unwrap(); 351 351 + let deserialized: ChunkedBlob = serde_json::from_str(&json).unwrap(); 352 352 + assert_eq!(deserialized.parts.len(), 2); 353 353 + assert_eq!(deserialized.total_size, 45000000); 213 354 } 214 355 }

+141 -29

src/yrs_pds.rs

reviewed

··· 62 62 pub fn doc_from_snapshot(data: &[u8]) -> Result<Doc, String> { 63 63 let doc = Doc::new(); 64 64 let _text = doc.get_or_insert_text("content"); 65 65 - let update = yrs::Update::decode_v1(data) 66 66 - .map_err(|e| format!("decode snapshot: {}", e))?; 65 65 + let update = yrs::Update::decode_v1(data).map_err(|e| format!("decode snapshot: {}", e))?; 67 66 doc.transact_mut() 68 67 .apply_update(update) 69 68 .map_err(|e| format!("apply snapshot: {}", e))?; ··· 72 71 73 72 /// Apply an incremental update to a Doc. 74 73 pub fn apply_update(doc: &Doc, data: &[u8]) -> Result<(), String> { 75 75 - let update = yrs::Update::decode_v1(data) 76 76 - .map_err(|e| format!("decode update: {}", e))?; 74 74 + let update = yrs::Update::decode_v1(data).map_err(|e| format!("decode update: {}", e))?; 77 75 doc.transact_mut() 78 76 .apply_update(update) 79 77 .map_err(|e| format!("apply update: {}", e))?; ··· 113 111 } 114 112 115 113 /// Reconstruct a Doc from a FileEntry by downloading blobs from PDS. 114 114 + /// 115 115 + /// If the entry has a pack_ref, extracts data from the pack blob. 116 116 + /// Otherwise downloads the snapshot blob directly. 116 117 pub async fn file_entry_to_doc( 117 118 entry: &FileEntry, 118 119 client: &PdsClient, 119 120 did: &str, 120 121 ) -> Result<Doc, String> { 121 121 - // Download snapshot blob 122 122 - let snapshot_data = client 123 123 - .get_blob(did, entry.snapshot_blob.cid()) 124 124 - .await?; 125 125 - 122 122 + let snapshot_data = get_file_blob_data(entry, client, did).await?; 126 123 let doc = doc_from_snapshot(&snapshot_data)?; 127 124 128 125 // Apply incremental updates if present ··· 134 131 Ok(doc) 135 132 } 136 133 134 134 + /// Get the raw blob data for a FileEntry, handling pack_ref extraction. 135 135 + pub async fn get_file_blob_data( 136 136 + entry: &FileEntry, 137 137 + client: &PdsClient, 138 138 + did: &str, 139 139 + ) -> Result<Vec<u8>, String> { 140 140 + if let Some(ref pack_ref) = entry.pack_ref { 141 141 + // Fetch pack blob, reassembling chunks if needed 142 142 + let pack_data = if let Some(ref chunks) = pack_ref.chunks { 143 143 + let mut chunk_data = Vec::new(); 144 144 + for chunk_ref in chunks { 145 145 + chunk_data.push(client.get_blob(did, chunk_ref.cid()).await?); 146 146 + } 147 147 + crate::pack::reassemble_chunks(&chunk_data) 148 148 + } else { 149 149 + client.get_blob(did, pack_ref.blob.cid()).await? 150 150 + }; 151 151 + let start = pack_ref.offset as usize; 152 152 + let end = start + pack_ref.length as usize; 153 153 + 154 154 + // Parse the pack (auto-detects gzip compression) 155 155 + let (_, blob_data) = crate::pack::parse_pack_auto(&pack_data)?; 156 156 + if end > blob_data.len() { 157 157 + return Err(format!( 158 158 + "pack_ref out of bounds: {}..{} in {} bytes", 159 159 + start, 160 160 + end, 161 161 + blob_data.len() 162 162 + )); 163 163 + } 164 164 + Ok(blob_data[start..end].to_vec()) 165 165 + } else { 166 166 + // Direct blob download 167 167 + client.get_blob(did, entry.snapshot_blob.cid()).await 168 168 + } 169 169 + } 170 170 + 137 171 /// Base64 encode bytes. 138 172 pub fn base64_encode(data: &[u8]) -> String { 139 173 use std::io::Write; ··· 153 187 let mut i = 0; 154 188 while i < chars.len() { 155 189 let a = decode_b64_char(chars[i])?; 156 156 - let b = if i + 1 < chars.len() { decode_b64_char(chars[i + 1])? } else { 0 }; 157 157 - let c = if i + 2 < chars.len() && chars[i + 2] != b'=' { decode_b64_char(chars[i + 2])? } else { 0 }; 158 158 - let d = if i + 3 < chars.len() && chars[i + 3] != b'=' { decode_b64_char(chars[i + 3])? } else { 0 }; 190 190 + let b = if i + 1 < chars.len() { 191 191 + decode_b64_char(chars[i + 1])? 192 192 + } else { 193 193 + 0 194 194 + }; 195 195 + let c = if i + 2 < chars.len() && chars[i + 2] != b'=' { 196 196 + decode_b64_char(chars[i + 2])? 197 197 + } else { 198 198 + 0 199 199 + }; 200 200 + let d = if i + 3 < chars.len() && chars[i + 3] != b'=' { 201 201 + decode_b64_char(chars[i + 3])? 202 202 + } else { 203 203 + 0 204 204 + }; 159 205 160 206 result.push((a << 2) | (b >> 4)); 161 207 if i + 2 < chars.len() && chars[i + 2] != b'=' { ··· 192 238 193 239 impl<'a> Base64Encoder<'a> { 194 240 fn new(buf: &'a mut Vec<u8>) -> Self { 195 195 - Self { buf, pending: [0; 3], pending_len: 0 } 241 241 + Self { 242 242 + buf, 243 243 + pending: [0; 3], 244 244 + pending_len: 0, 245 245 + } 196 246 } 197 247 198 248 fn finish(mut self) -> Result<(), std::io::Error> { ··· 204 254 205 255 fn flush_pending(&mut self) { 206 256 let a = self.pending[0]; 207 207 - let b = if self.pending_len > 1 { self.pending[1] } else { 0 }; 208 208 - let c = if self.pending_len > 2 { self.pending[2] } else { 0 }; 257 257 + let b = if self.pending_len > 1 { 258 258 + self.pending[1] 259 259 + } else { 260 260 + 0 261 261 + }; 262 262 + let c = if self.pending_len > 2 { 263 263 + self.pending[2] 264 264 + } else { 265 265 + 0 266 266 + }; 209 267 210 268 self.buf.push(B64_CHARS[(a >> 2) as usize]); 211 211 - self.buf.push(B64_CHARS[(((a & 0x3) << 4) | (b >> 4)) as usize]); 269 269 + self.buf 270 270 + .push(B64_CHARS[(((a & 0x3) << 4) | (b >> 4)) as usize]); 212 271 213 272 if self.pending_len > 1 { 214 214 - self.buf.push(B64_CHARS[(((b & 0xF) << 2) | (c >> 6)) as usize]); 273 273 + self.buf 274 274 + .push(B64_CHARS[(((b & 0xF) << 2) | (c >> 6)) as usize]); 215 275 } else { 216 276 self.buf.push(b'='); 217 277 } ··· 295 355 encode_snapshot(doc) 296 356 } 297 357 358 358 + /// Materialize manifest content as a string (for the content field). 359 359 + pub fn materialize_manifest_content(doc: &Doc) -> String { 360 360 + let entries = manifest_entries(doc); 361 361 + let mut lines: Vec<String> = entries 362 362 + .iter() 363 363 + .map(|(path, kind)| { 364 364 + let kind_str = match kind { 365 365 + FileKind::Text => "text", 366 366 + FileKind::Binary => "binary", 367 367 + }; 368 368 + format!("{}:{}", path, kind_str) 369 369 + }) 370 370 + .collect(); 371 371 + lines.sort(); 372 372 + lines.join("\n") 373 373 + } 374 374 + 298 375 /// Restore a manifest Doc from a snapshot. 299 376 pub fn manifest_from_snapshot(data: &[u8]) -> Result<Doc, String> { 300 377 let doc = Doc::new(); 301 378 let _map = doc.get_or_insert_map("manifest"); 302 302 - let update = yrs::Update::decode_v1(data) 303 303 - .map_err(|e| format!("decode manifest snapshot: {}", e))?; 379 379 + let update = 380 380 + yrs::Update::decode_v1(data).map_err(|e| format!("decode manifest snapshot: {}", e))?; 304 381 doc.transact_mut() 305 382 .apply_update(update) 306 383 .map_err(|e| format!("apply manifest snapshot: {}", e))?; ··· 320 397 pub fn is_text_extension(ext: &str) -> bool { 321 398 matches!( 322 399 ext.to_lowercase().as_str(), 323 323 - "md" | "html" | "css" | "js" | "ts" | "json" | "toml" | "yaml" | "yml" | "txt" | "xml" 324 324 - | "svg" | "jsx" | "tsx" | "rs" | "py" | "go" | "java" | "c" | "cpp" | "h" | "hpp" 325 325 - | "sh" | "bash" | "zsh" | "fish" | "rb" | "php" | "vue" | "svelte" | "astro" 326 326 - | "mdx" | "csv" | "ini" | "cfg" | "conf" | "env" | "gitignore" | "dockerignore" 327 327 - | "editorconfig" | "prettierrc" | "eslintrc" 400 400 + "md" | "html" 401 401 + | "css" 402 402 + | "js" 403 403 + | "ts" 404 404 + | "json" 405 405 + | "toml" 406 406 + | "yaml" 407 407 + | "yml" 408 408 + | "txt" 409 409 + | "xml" 410 410 + | "svg" 411 411 + | "jsx" 412 412 + | "tsx" 413 413 + | "rs" 414 414 + | "py" 415 415 + | "go" 416 416 + | "java" 417 417 + | "c" 418 418 + | "cpp" 419 419 + | "h" 420 420 + | "hpp" 421 421 + | "sh" 422 422 + | "bash" 423 423 + | "zsh" 424 424 + | "fish" 425 425 + | "rb" 426 426 + | "php" 427 427 + | "vue" 428 428 + | "svelte" 429 429 + | "astro" 430 430 + | "mdx" 431 431 + | "csv" 432 432 + | "ini" 433 433 + | "cfg" 434 434 + | "conf" 435 435 + | "env" 436 436 + | "gitignore" 437 437 + | "dockerignore" 438 438 + | "editorconfig" 439 439 + | "prettierrc" 440 440 + | "eslintrc" 328 441 ) 329 442 } 330 443 ··· 416 529 417 530 #[test] 418 531 fn manifest_crdt_merge_concurrent_add() { 419 419 - 420 420 - 421 532 // Two sites start from same base 422 533 let base = new_manifest_doc(); 423 534 manifest_insert(&base, "shared.md", &FileKind::Text); ··· 447 558 448 559 #[test] 449 560 fn manifest_crdt_merge_set_wins_over_delete() { 450 450 - 451 451 - 452 561 // Base has a file 453 562 let base = new_manifest_doc(); 454 563 manifest_insert(&base, "file.md", &FileKind::Text); ··· 470 579 } 471 580 472 581 let entries = manifest_entries(&site_a); 473 473 - assert!(entries.contains_key("file.md"), "edit wins: set should win over delete"); 582 582 + assert!( 583 583 + entries.contains_key("file.md"), 584 584 + "edit wins: set should win over delete" 585 585 + ); 474 586 } 475 587 476 588 #[test]

+4 -13

tests/e2e_tests.rs

reviewed

··· 171 171 result.files_uploaded, 1, 172 172 "only changed file should be uploaded" 173 173 ); 174 174 - assert_eq!( 175 175 - result.files_skipped, 2, 176 176 - "unchanged files should be skipped" 177 177 - ); 174 174 + assert_eq!(result.files_skipped, 2, "unchanged files should be skipped"); 178 175 179 176 // Verify content 180 177 let dest = tempfile::tempdir().unwrap(); ··· 251 248 252 249 // Merge both 253 250 let merged = tempfile::tempdir().unwrap(); 254 254 - pds_yrs::merge_sites( 255 255 - &client, 256 256 - &did, 257 257 - &[&rkey_a, &rkey_b], 258 258 - merged.path(), 259 259 - false, 260 260 - ) 261 261 - .await 262 262 - .unwrap(); 251 251 + pds_yrs::merge_sites(&client, &did, &[&rkey_a, &rkey_b], merged.path(), false) 252 252 + .await 253 253 + .unwrap(); 263 254 264 255 let content = read_file(merged.path(), "shared.md").await; 265 256 // Note: since these are independent docs with different client IDs,