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

syncweb_core/node/
blob_store.rs
