syncweb_core/node/
blob_store.rs

1use bytes::Bytes;
2use iroh::{EndpointAddr, address_lookup::memory::MemoryLookup, endpoint::Endpoint};
3use iroh_blobs::{
4    BlobFormat, BlobsProtocol, Hash,
5    api::{
6        Store as BlobApi,
7        blobs::{AddPathOptions, BlobReader, BlobStatus, ExportMode, ExportOptions, ImportMode},
8    },
9    protocol::GetRequest,
10    ticket::BlobTicket,
11};
12use n0_future::StreamExt;
13use std::path::{Path, PathBuf};
14
15use crate::error::{Result, SyncwebError};
16use crate::net::NetworkContext;
17
18#[derive(Clone)]
19pub struct BlobStore {
20    store: BlobApi,
21    address_lookup: MemoryLookup,
22}
23
24impl BlobStore {
25    #[must_use]
26    pub fn new(protocol: &BlobsProtocol) -> Self {
27        Self::new_with_address_lookup(protocol, MemoryLookup::new())
28    }
29
30    #[must_use]
31    pub fn new_with_address_lookup(protocol: &BlobsProtocol, address_lookup: MemoryLookup) -> Self {
32        Self {
33            store: protocol.store().clone(),
34            address_lookup,
35        }
36    }
37
38    #[must_use]
39    pub const fn inner(&self) -> &BlobApi {
40        &self.store
41    }
42
43    /// Create a reader for the given hash.
44    ///
45    /// The reader implements [`tokio::io::AsyncRead`] and [`tokio::io::AsyncSeek`],
46    /// allowing O(1)-memory streaming to any sink.
47    #[must_use]
48    pub fn reader(&self, hash: impl Into<Hash>) -> BlobReader {
49        self.store.reader(hash)
50    }
51
52    /// # Errors
53    ///
54    /// Returns an error if the blob fails to be added to the store.
55    pub async fn add_bytes(&self, data: impl AsRef<[u8]>) -> Result<Hash> {
56        Ok(self
57            .store
58            .add_bytes(Bytes::copy_from_slice(data.as_ref()))
59            .await
60            .map_err(|error| SyncwebError::operation("failed to add blob bytes", error))?
61            .hash)
62    }
63
64    /// # Errors
65    ///
66    /// Returns an error if the file fails to be read or added to the store.
67    pub async fn add_file(&self, path: impl AsRef<Path>) -> Result<Hash> {
68        let path = absolute_path(path)?;
69        Ok(self
70            .store
71            .add_path(path)
72            .await
73            .map_err(|error| SyncwebError::operation("failed to add blob file", error))?
74            .hash)
75    }
76
77    /// Add a file using a reference to its original path instead of copying
78    /// its contents into the blob store.
79    ///
80    /// The source file must remain available and unchanged for as long as the
81    /// blob store may need to read this blob.
82    ///
83    /// # Errors
84    ///
85    /// Returns an error if the file fails to be read or added to the store.
86    pub async fn add_file_ref(&self, path: impl AsRef<Path>) -> Result<Hash> {
87        let path = absolute_path(path)?;
88        Ok(self
89            .store
90            .add_path_with_opts(AddPathOptions {
91                path,
92                mode: ImportMode::TryReference,
93                format: BlobFormat::Raw,
94            })
95            .await
96            .map_err(|error| SyncwebError::operation("failed to add blob file (reference)", error))?
97            .hash)
98    }
99
100    /// Export a complete blob to a temporary or caller-managed path.
101    ///
102    /// Uses [`ExportMode::Copy`] — the blob stays in the store and a copy
103    /// (or `CoW` reflink) is written to `destination`.
104    ///
105    /// # Errors
106    ///
107    /// Returns an error if the blob cannot be read or written.
108    pub async fn export_to_path(&self, hash: Hash, destination: impl AsRef<Path>) -> Result<u64> {
109        self.export_to_path_with_mode(hash, destination, ExportMode::Copy).await
110    }
111
112    /// Export a complete blob with a specific [`ExportMode`].
113    ///
114    /// `ExportMode::TryReference` moves the blob's internal file to
115    /// `destination` and updates the store to track it externally,
116    /// avoiding a copy. `ExportMode::Copy` leaves the blob in the store
117    /// and writes a copy (or `CoW` reflink) to `destination`.
118    ///
119    /// # Errors
120    ///
121    /// Returns an error if the blob cannot be read or written.
122    pub async fn export_to_path_with_mode(
123        &self,
124        hash: Hash,
125        destination: impl AsRef<Path>,
126        mode: ExportMode,
127    ) -> Result<u64> {
128        self.store
129            .export_with_opts(ExportOptions {
130                hash,
131                mode,
132                target: destination.as_ref().to_owned(),
133            })
134            .await
135            .map_err(|error| SyncwebError::operation("failed to export blob", error))
136    }
137
138    /// # Errors
139    ///
140    /// Returns an error if the store cannot be queried.
141    pub async fn has(&self, hash: Hash) -> Result<bool> {
142        self.store
143            .has(hash)
144            .await
145            .map_err(|error| SyncwebError::operation("failed to query blob store", error))
146    }
147
148    /// Get metadata (size) for a blob.
149    ///
150    /// # Errors
151    ///
152    /// Returns an error if the blob is not found or the store cannot be queried.
153    pub async fn stat(&self, hash: Hash) -> Result<BlobStatus> {
154        self.store
155            .blobs()
156            .status(hash)
157            .await
158            .map_err(|error| SyncwebError::operation("failed to stat blob", error))
159    }
160
161    /// # Errors
162    ///
163    /// Returns an error if the blob cannot be found or read.
164    pub async fn get(&self, hash: Hash) -> Result<Bytes> {
165        self.store
166            .get_bytes(hash)
167            .await
168            .map_err(|error| SyncwebError::operation("failed to read blob", error))
169    }
170
171    #[must_use]
172    pub fn ticket(&self, endpoint: &Endpoint, hash: Hash) -> BlobTicket {
173        self.ticket_for_addr(endpoint.addr(), hash)
174    }
175
176    #[must_use]
177    pub fn ticket_for_addr(&self, addr: EndpointAddr, hash: Hash) -> BlobTicket {
178        BlobTicket::new(addr, hash, iroh_blobs::BlobFormat::Raw)
179    }
180
181    /// Pin a blob with a durable named tag so garbage collection cannot remove it.
182    ///
183    /// # Errors
184    ///
185    /// Returns an error if the tag cannot be written.
186    pub async fn pin(&self, name: impl AsRef<str>, hash: Hash) -> Result<()> {
187        self.store
188            .tags()
189            .set(name.as_ref(), hash)
190            .await
191            .map_err(|error| SyncwebError::operation("failed to pin blob", error))
192    }
193
194    /// Remove a durable pin from a blob.
195    ///
196    /// # Errors
197    ///
198    /// Returns an error if the tag cannot be removed.
199    pub async fn unpin(&self, name: impl AsRef<str>) -> Result<()> {
200        self.store
201            .tags()
202            .delete(name.as_ref())
203            .await
204            .map(|_deleted| ())
205            .map_err(|error| SyncwebError::operation("failed to unpin blob", error))
206    }
207
208    /// # Errors
209    ///
210    /// Returns an error if the pin cannot be queried.
211    pub async fn is_pinned(&self, name: impl AsRef<str>, hash: Hash) -> Result<bool> {
212        self.store
213            .tags()
214            .get(name.as_ref())
215            .await
216            .map(|tag_info| tag_info.is_some_and(|tag| tag.hash == hash))
217            .map_err(|error| SyncwebError::operation("failed to query blob pin", error))
218    }
219
220    /// List all complete blobs currently in the store.
221    ///
222    /// # Errors
223    ///
224    /// Returns an error if the store cannot be queried.
225    pub async fn list_hashes(&self) -> Result<Vec<Hash>> {
226        self.store
227            .blobs()
228            .list()
229            .hashes()
230            .await
231            .map_err(|error| SyncwebError::operation("failed to list blobs", error))
232    }
233
234    /// List named pins whose names start with `prefix`.
235    ///
236    /// # Errors
237    ///
238    /// Returns an error if the store cannot be queried.
239    pub async fn list_pins(&self, prefix: impl AsRef<[u8]>) -> Result<Vec<(String, Hash)>> {
240        let mut tags = self
241            .store
242            .tags()
243            .list_prefix(prefix)
244            .await
245            .map_err(|error| SyncwebError::operation("failed to list blob pins", error))?;
246        let mut pins = Vec::new();
247        while let Some(tag_result) = tags.next().await {
248            let tag = tag_result.map_err(|error| SyncwebError::operation("failed to read blob pin", error))?;
249            let name = String::from_utf8(tag.name.as_ref().to_vec())
250                .map_err(|error| SyncwebError::operation("blob pin name is not UTF-8", error))?;
251            pins.push((name, tag.hash));
252        }
253        Ok(pins)
254    }
255
256    /// # Errors
257    ///
258    /// Returns an error if the blob cannot be fetched from the remote endpoint.
259    pub async fn fetch(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<()> {
260        self.address_lookup.add_endpoint_info(ticket.addr().clone());
261        self.store
262            .downloader(endpoint)
263            .download(ticket.hash_and_format(), [ticket.addr().id])
264            .await
265            .map_err(|error| SyncwebError::operation("failed to fetch blob", error))?;
266        Ok(())
267    }
268
269    /// Fetch a blob with network access control gating.
270    ///
271    /// Checks via `NetworkContext` that the local node can access the blob
272    /// through at least one network membership before downloading.
273    ///
274    /// # Errors
275    ///
276    /// Returns `AccessDenied` if the blob is not associated with any network
277    /// the local node belongs to. Returns other errors if the fetch itself fails.
278    pub async fn download_with_network(
279        &self,
280        endpoint: &Endpoint,
281        ticket: &BlobTicket,
282        ctx: &NetworkContext,
283    ) -> Result<()> {
284        let hash = ticket.hash();
285        if !ctx.can_access_blob(&hash)? {
286            return Err(SyncwebError::access_denied("blob not accessible in any network"));
287        }
288        self.fetch(endpoint, ticket).await
289    }
290
291    /// Force-fetch a blob from a remote peer, re-downloading even if the blob
292    /// is already present locally. Uses `iroh-blobs`' low-level `execute_get`
293    /// API which does not check local availability.
294    ///
295    /// # Errors
296    ///
297    /// Returns an error if the connection or download fails.
298    pub async fn force_fetch(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<()> {
299        self.address_lookup.add_endpoint_info(ticket.addr().clone());
300        let hash = ticket.hash();
301        let conn = endpoint
302            .connect(ticket.addr().id, iroh_blobs::protocol::ALPN)
303            .await
304            .map_err(|error| SyncwebError::operation("failed to connect to provider", error))?;
305        self.force_fetch_from_conn(conn, hash).await
306    }
307
308    /// Force-fetch a blob from a known peer, connecting by public key alone.
309    ///
310    /// # Errors
311    ///
312    /// Returns an error if the connection or download fails.
313    pub async fn force_fetch_from_peer(&self, endpoint: &Endpoint, peer: &iroh::PublicKey, hash: Hash) -> Result<()> {
314        let conn = endpoint
315            .connect(*peer, iroh_blobs::protocol::ALPN)
316            .await
317            .map_err(|error| SyncwebError::operation("failed to connect to peer", error))?;
318        self.force_fetch_from_conn(conn, hash).await
319    }
320
321    async fn force_fetch_from_conn(&self, conn: iroh::endpoint::Connection, hash: Hash) -> Result<()> {
322        let request = GetRequest::blob(hash);
323        self.store
324            .remote()
325            .execute_get(conn, request)
326            .await
327            .map_err(|error| SyncwebError::operation("failed to force-fetch blob", error))?;
328        Ok(())
329    }
330}
331
332fn absolute_path(path: impl AsRef<Path>) -> Result<PathBuf> {
333    std::path::absolute(path.as_ref())
334        .map_err(|error| SyncwebError::operation("failed to resolve blob file path", error))
335}
syncweb_core/node/blob_store.rs

syncweb_core/node/
blob_store.rs
