1use anyhow::{Context, Result};
2use futures::stream::{self, StreamExt};
3use std::collections::HashSet;
4use std::path::Path;
5use std::sync::Arc;
6
7use crate::config::ensure_keys_string;
8use crate::fetch::{FetchConfig, Fetcher};
9use crate::HashtreeStore;
10use hashtree_core::{to_hex, Cid, HashTree, HashTreeConfig, Link};
11
12const BLOSSOM_PUSH_CONCURRENCY: usize = 16;
13const BLOSSOM_PUSH_PROGRESS_EVERY: usize = 512;
14
15fn parse_root_cid(cid_str: &str) -> Result<Cid> {
16 Cid::parse(cid_str).map_err(|e| anyhow::anyhow!("Invalid CID '{}': {}", cid_str, e))
17}
18
19fn child_cid(parent: &Cid, link: &Link) -> Cid {
20 let inherits_parent_key = link
21 .name
22 .as_deref()
23 .map(|name| {
24 name.starts_with("_chunk_")
25 || (name.starts_with('_') && name.chars().count() == 2 && link.link_type.is_tree())
26 })
27 .unwrap_or(false);
28
29 Cid {
30 hash: link.hash,
31 key: link.key.or(if inherits_parent_key {
32 parent.key
33 } else {
34 None
35 }),
36 }
37}
38
39async fn ensure_local_blob_for_push(
40 store: &HashtreeStore,
41 fetcher: Option<&Fetcher>,
42 cid: &Cid,
43) -> Result<()> {
44 if store.get_blob(&cid.hash)?.is_some() {
45 return Ok(());
46 }
47
48 if let Some(fetcher) = fetcher {
49 let data = fetcher
50 .fetch_chunk(None, &to_hex(&cid.hash))
51 .await
52 .with_context(|| format!("failed to hydrate missing local blob {}", cid))?;
53 store
54 .put_blob(&data)
55 .with_context(|| format!("failed to persist hydrated blob {}", cid))?;
56 if store.get_blob(&cid.hash)?.is_some() {
57 return Ok(());
58 }
59 }
60
61 anyhow::bail!("missing local blob while pushing DAG: {}", cid);
62}
63
64pub(crate) async fn collect_cids_for_push(
65 store: &HashtreeStore,
66 root_cid: Cid,
67 fetcher: Option<&Fetcher>,
68) -> Result<Vec<Cid>> {
69 let mut cids_to_push = Vec::new();
70 let mut visited: HashSet<[u8; 32]> = HashSet::new();
71 let mut queue = vec![root_cid];
72 let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
73
74 while let Some(cid) = queue.pop() {
75 if !visited.insert(cid.hash) {
76 continue;
77 }
78
79 ensure_local_blob_for_push(store, fetcher, &cid).await?;
80 cids_to_push.push(cid.clone());
81
82 let node = tree
83 .get_node(&cid)
84 .await
85 .map_err(|e| anyhow::anyhow!("Failed to inspect {}: {}", cid, e))?;
86
87 if let Some(node) = node {
88 for link in &node.links {
89 if !visited.contains(&link.hash) {
90 queue.push(child_cid(&cid, link));
91 }
92 }
93 }
94 }
95
96 Ok(cids_to_push)
97}
98
99fn matching_old_child<'a>(
100 old_links: &'a [Link],
101 new_index: usize,
102 new_link: &Link,
103) -> Option<&'a Link> {
104 if let Some(name) = new_link.name.as_deref() {
105 old_links
106 .iter()
107 .find(|old_link| old_link.name.as_deref() == Some(name))
108 } else {
109 old_links
110 .get(new_index)
111 .filter(|old_link| old_link.name.is_none())
112 }
113}
114
115pub(crate) async fn collect_incremental_cids_for_push(
116 store: &HashtreeStore,
117 root_cid: Cid,
118 previous_root_cid: Cid,
119 fetcher: Option<&Fetcher>,
120) -> Result<Vec<Cid>> {
121 let mut cids_to_push = Vec::new();
122 let mut visited_new: HashSet<[u8; 32]> = HashSet::new();
123 let mut queue = vec![(root_cid, Some(previous_root_cid))];
124 let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
125
126 while let Some((cid, old_cid)) = queue.pop() {
127 if old_cid.as_ref().is_some_and(|old| old.hash == cid.hash) {
128 continue;
129 }
130 if !visited_new.insert(cid.hash) {
131 continue;
132 }
133
134 ensure_local_blob_for_push(store, fetcher, &cid).await?;
135 cids_to_push.push(cid.clone());
136
137 let node = tree
138 .get_node(&cid)
139 .await
140 .map_err(|e| anyhow::anyhow!("Failed to inspect {}: {}", cid, e))?;
141 let Some(node) = node else {
142 continue;
143 };
144
145 let old_node = match old_cid.as_ref() {
146 Some(old_cid) => match tree.get_node(old_cid).await {
147 Ok(old_node) => old_node,
148 Err(err) => {
149 tracing::warn!(
150 "Failed to inspect previous Blossom DAG node {}; uploading changed subtree: {}",
151 old_cid,
152 err
153 );
154 None
155 }
156 },
157 None => None,
158 };
159
160 for (index, link) in node.links.iter().enumerate() {
161 let child = child_cid(&cid, link);
162 let old_child = old_node
163 .as_ref()
164 .and_then(|old_node| matching_old_child(&old_node.links, index, link))
165 .map(|old_link| child_cid(old_cid.as_ref().expect("old node has cid"), old_link));
166
167 if old_child
168 .as_ref()
169 .is_some_and(|old_child| old_child.hash == child.hash)
170 {
171 continue;
172 }
173 queue.push((child, old_child));
174 }
175 }
176
177 Ok(cids_to_push)
178}
179
180async fn upload_cids_with_client(
181 store: Arc<HashtreeStore>,
182 fetcher: Option<Arc<Fetcher>>,
183 client: hashtree_blossom::BlossomClient,
184 cids_to_push: Vec<Cid>,
185 force_upload: bool,
186) -> Result<(usize, usize, usize, Option<String>)> {
187 let total = cids_to_push.len();
188 let mut total_uploaded = 0usize;
189 let mut total_skipped = 0usize;
190 let mut total_errors = 0usize;
191 let mut last_error = None;
192 let mut processed = 0usize;
193
194 let mut uploads = stream::iter(cids_to_push.into_iter().map(|cid| {
195 let store = Arc::clone(&store);
196 let fetcher = fetcher.clone();
197 let client = client.clone();
198 async move {
199 ensure_local_blob_for_push(store.as_ref(), fetcher.as_deref(), &cid).await?;
200 let data = store
201 .get_blob(&cid.hash)?
202 .ok_or_else(|| anyhow::anyhow!("missing local blob while uploading {}", cid))?;
203 if force_upload {
204 client
205 .upload_to_selected_servers(&data, client.write_servers())
206 .await
207 .map(|(hash, _successes)| (hash, true))
208 .map_err(|error| anyhow::anyhow!(error.to_string()))
209 } else {
210 client
211 .upload_if_missing(&data)
212 .await
213 .map_err(|error| anyhow::anyhow!(error.to_string()))
214 }
215 }
216 }))
217 .buffer_unordered(BLOSSOM_PUSH_CONCURRENCY);
218
219 while let Some(result) = uploads.next().await {
220 processed += 1;
221 match result {
222 Ok((_hash, was_uploaded)) => {
223 if was_uploaded {
224 total_uploaded += 1;
225 } else {
226 total_skipped += 1;
227 }
228 }
229 Err(error) => {
230 tracing::warn!("Blossom upload failed: {}", error);
231 total_errors += 1;
232 last_error = Some(error.to_string());
233 }
234 }
235
236 if processed % BLOSSOM_PUSH_PROGRESS_EVERY == 0 || processed == total {
237 println!(
238 " file servers: {processed}/{total} processed ({total_uploaded} uploaded, {total_skipped} already exist, {total_errors} failed)",
239 );
240 }
241 }
242
243 Ok((total_uploaded, total_skipped, total_errors, last_error))
244}
245
246pub async fn push_to_blossom(
248 data_dir: &Path,
249 cid_str: &str,
250 server_override: Option<String>,
251 force_upload: bool,
252 shallow: bool,
253) -> Result<()> {
254 use hashtree_blossom::BlossomClient;
255 use nostr::Keys;
256
257 let (nsec_str, _) = ensure_keys_string()?;
258 let keys = Keys::parse(&nsec_str).context("Failed to parse nsec")?;
259
260 let client = if let Some(server) = server_override {
261 BlossomClient::new(keys).with_write_servers(vec![server])
262 } else {
263 BlossomClient::new(keys)
264 };
265
266 if client.write_servers().is_empty() {
267 anyhow::bail!(
268 "No file servers configured. Use --server or add write_servers to config.toml"
269 );
270 }
271
272 let store = Arc::new(HashtreeStore::new(data_dir)?);
273 let fetcher = Arc::new(Fetcher::new(FetchConfig::default()));
274
275 let root_cid = parse_root_cid(cid_str)?;
276 let cids_to_push = if shallow {
277 ensure_local_blob_for_push(store.as_ref(), Some(fetcher.as_ref()), &root_cid).await?;
278 vec![root_cid]
279 } else {
280 println!("Collecting blocks...");
281 collect_cids_for_push(store.as_ref(), root_cid, Some(fetcher.as_ref())).await?
282 };
283
284 println!("Found {} blocks to push", cids_to_push.len());
285 let (uploaded, skipped, errors, last_error) =
286 upload_cids_with_client(store, Some(fetcher), client, cids_to_push, force_upload).await?;
287
288 println!(
289 "\nUploaded: {}, Skipped: {}, Errors: {}",
290 uploaded, skipped, errors
291 );
292 if let Some(last_error) = last_error {
293 eprintln!("Last error: {last_error}");
294 }
295 println!("Done!");
296 Ok(())
297}
298
299pub async fn background_blossom_push(
301 data_dir: &Path,
302 cid_str: &str,
303 servers: &[String],
304) -> Result<()> {
305 let store = Arc::new(HashtreeStore::new(data_dir)?);
306 background_blossom_push_with_store(store, cid_str, servers).await
307}
308
309pub async fn background_blossom_push_with_store(
310 store: Arc<HashtreeStore>,
311 cid_str: &str,
312 servers: &[String],
313) -> Result<()> {
314 let root_cid = parse_root_cid(cid_str)?;
315 background_blossom_push_incremental_with_store(store, root_cid, None, servers).await
316}
317
318pub async fn background_blossom_push_incremental_with_store(
319 store: Arc<HashtreeStore>,
320 root_cid: Cid,
321 previous_root_cid: Option<Cid>,
322 servers: &[String],
323) -> Result<()> {
324 use hashtree_blossom::BlossomClient;
325 use nostr::Keys;
326
327 let (nsec_str, _) = ensure_keys_string()?;
328 let keys = Keys::parse(&nsec_str).context("Failed to parse nsec")?;
329
330 let fetcher = Arc::new(Fetcher::new(FetchConfig::default()));
331 let cids_to_push = if let Some(previous_root_cid) = previous_root_cid.as_ref() {
332 println!("Collecting bounded DAG diff for file-server push...");
333 match collect_incremental_cids_for_push(
334 store.as_ref(),
335 root_cid.clone(),
336 previous_root_cid.clone(),
337 Some(fetcher.as_ref()),
338 )
339 .await
340 {
341 Ok(cids) => cids,
342 Err(err) => {
343 tracing::warn!(
344 "Blossom DAG diff failed; falling back to full push: {}",
345 err
346 );
347 collect_cids_for_push(store.as_ref(), root_cid, Some(fetcher.as_ref())).await?
348 }
349 }
350 } else {
351 println!("Collecting DAG for file-server push...");
352 collect_cids_for_push(store.as_ref(), root_cid, Some(fetcher.as_ref())).await?
353 };
354
355 if cids_to_push.is_empty() {
356 return Ok(());
357 }
358
359 let client = if servers.is_empty() {
360 BlossomClient::new(keys)
361 } else {
362 BlossomClient::new(keys).with_write_servers(servers.to_vec())
363 };
364 let (_total_uploaded, _total_skipped, total_errors, last_error) =
365 upload_cids_with_client(store, Some(fetcher), client, cids_to_push, false).await?;
366
367 if total_errors > 0 {
368 let detail = last_error
369 .as_deref()
370 .map(|err| format!(" (last error: {err})"))
371 .unwrap_or_default();
372 anyhow::bail!(
373 "failed to upload {} blob(s) to configured file servers{}",
374 total_errors,
375 detail
376 );
377 }
378
379 Ok(())
380}
381
382#[cfg(test)]
383mod tests {
384 use super::{collect_cids_for_push, collect_incremental_cids_for_push};
385 use crate::HashtreeStore;
386 use futures::executor::block_on as sync_block_on;
387 use hashtree_core::{DirEntry, HashTree, HashTreeConfig, LinkType};
388 use tempfile::tempdir;
389
390 #[tokio::test]
391 async fn collect_cids_for_push_fails_on_missing_descendant_blob() {
392 let tmp = tempdir().expect("tempdir");
393 let store = HashtreeStore::with_options(tmp.path(), None, 32 * 1024 * 1024).expect("store");
394 let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
395
396 let root = sync_block_on(async {
397 let (file_cid, _size) = tree.put_file(b"hello").await.expect("file");
398 tree.put_directory(vec![hashtree_core::DirEntry::from_cid(
399 "greeting.txt",
400 &file_cid,
401 )])
402 .await
403 .expect("dir")
404 });
405
406 let entries = store
407 .get_tree_node(&root.hash)
408 .expect("root node")
409 .expect("root node present")
410 .links;
411 let child_hash = entries[0].hash;
412 store
413 .router()
414 .delete_local_only(&child_hash)
415 .expect("delete child locally");
416
417 let err = collect_cids_for_push(&store, root, None)
418 .await
419 .expect_err("missing child should fail");
420 assert!(err
421 .to_string()
422 .contains("missing local blob while pushing DAG"));
423 }
424
425 #[tokio::test]
426 async fn incremental_push_collects_only_changed_named_subtrees() {
427 let tmp = tempdir().expect("tempdir");
428 let store = HashtreeStore::with_options(tmp.path(), None, 32 * 1024 * 1024).expect("store");
429 let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
430
431 let stable_file = tree.put_blob(b"stable").await.expect("stable file");
432 let old_changed_file = tree.put_blob(b"old").await.expect("old file");
433 let old_subdir = tree
434 .put_directory(vec![
435 DirEntry::new("changed.txt", old_changed_file).with_size(3),
436 DirEntry::new("stable.txt", stable_file).with_size(6),
437 ])
438 .await
439 .expect("old subdir");
440 let old_root = tree
441 .put_directory(vec![
442 DirEntry::new("subdir", old_subdir.hash).with_link_type(LinkType::Dir),
443 DirEntry::new("stable-root.txt", stable_file).with_size(6),
444 ])
445 .await
446 .expect("old root");
447
448 let new_changed_file = tree.put_blob(b"new").await.expect("new file");
449 let new_subdir = tree
450 .put_directory(vec![
451 DirEntry::new("stable.txt", stable_file).with_size(6),
452 DirEntry::new("changed.txt", new_changed_file).with_size(3),
453 ])
454 .await
455 .expect("new subdir");
456 let new_root = tree
457 .put_directory(vec![
458 DirEntry::new("stable-root.txt", stable_file).with_size(6),
459 DirEntry::new("subdir", new_subdir.hash).with_link_type(LinkType::Dir),
460 ])
461 .await
462 .expect("new root");
463
464 let cids = collect_incremental_cids_for_push(&store, new_root.clone(), old_root, None)
465 .await
466 .expect("incremental cids");
467 let hashes = cids.iter().map(|cid| cid.hash).collect::<Vec<_>>();
468
469 assert_eq!(hashes.len(), 3);
470 assert!(hashes.contains(&new_root.hash));
471 assert!(hashes.contains(&new_subdir.hash));
472 assert!(hashes.contains(&new_changed_file));
473 assert!(!hashes.contains(&stable_file));
474 }
475}