syncweb_core/sync/
lazy_fetch.rs

1use std::path::Path;
2
3use iroh::endpoint::Endpoint;
4use iroh_blobs::Hash;
5use iroh_blobs::ticket::BlobTicket;
6
7use crate::{
8    error::Result,
9    node::{blob_store::BlobStore, docs_engine::DocsEngine},
10    sync::{IntentHandle, SyncEvent},
11};
12
13/// On-demand content fetcher for selective sync.
14#[derive(Clone)]
15pub struct LazyFetch {
16    blob_store: BlobStore,
17    _docs_engine: DocsEngine,
18}
19
20impl LazyFetch {
21    #[must_use]
22    pub const fn new(blob_store: BlobStore, docs_engine: DocsEngine) -> Self {
23        Self {
24            blob_store,
25            _docs_engine: docs_engine,
26        }
27    }
28
29    /// Fetch a blob only when the caller asks for it.
30    /// # Errors
31    ///
32    /// Returns an error if the database cannot be accessed.
33    pub async fn fetch(&self, hash: Hash) -> Result<bytes::Bytes> {
34        self.blob_store.get(hash).await
35    }
36
37    /// Fetch a missing blob from a peer ticket, then read it locally.
38    /// # Errors
39    ///
40    /// Returns an error if the network or database cannot be accessed.
41    pub async fn fetch_remote(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<bytes::Bytes> {
42        self.blob_store.fetch(endpoint, ticket).await?;
43        self.fetch(ticket.hash()).await
44    }
45
46    /// Fetch and write a blob to a local path.
47    /// # Errors
48    ///
49    /// Returns an error if the filesystem or database cannot be accessed.
50    pub async fn fetch_to(&self, hash: Hash, path: impl AsRef<Path>) -> Result<()> {
51        let bytes = self.fetch(hash).await?;
52        if let Some(parent) = path.as_ref().parent() {
53            tokio::fs::create_dir_all(parent).await?;
54        }
55        tokio::fs::write(path, bytes).await?;
56        Ok(())
57    }
58
59    /// Create an intent that reports the local fetch lifecycle.
60    #[must_use]
61    pub fn fetch_intent(&self, hash: Hash) -> IntentHandle {
62        let (events, mut commands, handle) = IntentHandle::channel();
63        let store = self.blob_store.clone();
64        tokio::spawn(async move {
65            let _ = events.send(SyncEvent::Started);
66            let mut paused = false;
67            let mut commands_open = true;
68            loop {
69                tokio::select! {
70                    result = store.get(hash), if !paused => {
71                        match result {
72                            Ok(bytes) => {
73                                let size = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
74                                let _ = events.send(SyncEvent::Progress { completed: size, total: Some(size) });
75                                let _ = events.send(SyncEvent::Finished);
76                            }
77                            Err(error) => { let _ = events.send(SyncEvent::Failed(error.to_string())); }
78                        }
79                        break;
80                    }
81                    command = commands.recv(), if commands_open => {
82                        match command {
83                            Some(crate::sync::SyncCommand::Pause) if !paused => {
84                                paused = true;
85                                let _ = events.send(SyncEvent::Paused);
86                            }
87                            Some(crate::sync::SyncCommand::Resume) if paused => {
88                                paused = false;
89                                let _ = events.send(SyncEvent::Resumed);
90                            }
91                            Some(crate::sync::SyncCommand::Cancel) => {
92                                let _ = events.send(SyncEvent::Cancelled);
93                                break;
94                            }
95                            Some(_) => {}
96                            None => {
97                                commands_open = false;
98                                paused = false;
99                            }
100                        }
101                    }
102                }
103            }
104        });
105        handle
106    }
107}
syncweb_core/sync/lazy_fetch.rs

syncweb_core/sync/
lazy_fetch.rs
