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;
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        Ok(self
69            .store
70            .add_path(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        Ok(self
87            .store
88            .add_path_with_opts(AddPathOptions {
89                path: path.as_ref().to_owned(),
90                mode: ImportMode::TryReference,
91                format: BlobFormat::Raw,
92            })
93            .await
94            .map_err(|error| SyncwebError::operation("failed to add blob file (reference)", error))?
95            .hash)
96    }
97
98    /// Export a complete blob to a temporary or caller-managed path.
99    ///
100    /// Uses [`ExportMode::Copy`] — the blob stays in the store and a copy
101    /// (or `CoW` reflink) is written to `destination`.
102    ///
103    /// # Errors
104    ///
105    /// Returns an error if the blob cannot be read or written.
106    pub async fn export_to_path(&self, hash: Hash, destination: impl AsRef<Path>) -> Result<u64> {
107        self.export_to_path_with_mode(hash, destination, ExportMode::Copy).await
108    }
109
110    /// Export a complete blob with a specific [`ExportMode`].
111    ///
112    /// `ExportMode::TryReference` moves the blob's internal file to
113    /// `destination` and updates the store to track it externally,
114    /// avoiding a copy. `ExportMode::Copy` leaves the blob in the store
115    /// and writes a copy (or `CoW` reflink) to `destination`.
116    ///
117    /// # Errors
118    ///
119    /// Returns an error if the blob cannot be read or written.
120    pub async fn export_to_path_with_mode(
121        &self,
122        hash: Hash,
123        destination: impl AsRef<Path>,
124        mode: ExportMode,
125    ) -> Result<u64> {
126        self.store
127            .export_with_opts(ExportOptions {
128                hash,
129                mode,
130                target: destination.as_ref().to_owned(),
131            })
132            .await
133            .map_err(|error| SyncwebError::operation("failed to export blob", error))
134    }
135
136    /// # Errors
137    ///
138    /// Returns an error if the store cannot be queried.
139    pub async fn has(&self, hash: Hash) -> Result<bool> {
140        self.store
141            .has(hash)
142            .await
143            .map_err(|error| SyncwebError::operation("failed to query blob store", error))
144    }
145
146    /// Get metadata (size) for a blob.
147    ///
148    /// # Errors
149    ///
150    /// Returns an error if the blob is not found or the store cannot be queried.
151    pub async fn stat(&self, hash: Hash) -> Result<BlobStatus> {
152        self.store
153            .blobs()
154            .status(hash)
155            .await
156            .map_err(|error| SyncwebError::operation("failed to stat blob", error))
157    }
158
159    /// # Errors
160    ///
161    /// Returns an error if the blob cannot be found or read.
162    pub async fn get(&self, hash: Hash) -> Result<Bytes> {
163        self.store
164            .get_bytes(hash)
165            .await
166            .map_err(|error| SyncwebError::operation("failed to read blob", error))
167    }
168
169    #[must_use]
170    pub fn ticket(&self, endpoint: &Endpoint, hash: Hash) -> BlobTicket {
171        self.ticket_for_addr(endpoint.addr(), hash)
172    }
173
174    #[must_use]
175    pub fn ticket_for_addr(&self, addr: EndpointAddr, hash: Hash) -> BlobTicket {
176        BlobTicket::new(addr, hash, iroh_blobs::BlobFormat::Raw)
177    }
178
179    /// Pin a blob with a durable named tag so garbage collection cannot remove it.
180    ///
181    /// # Errors
182    ///
183    /// Returns an error if the tag cannot be written.
184    pub async fn pin(&self, name: impl AsRef<str>, hash: Hash) -> Result<()> {
185        self.store
186            .tags()
187            .set(name.as_ref(), hash)
188            .await
189            .map_err(|error| SyncwebError::operation("failed to pin blob", error))
190    }
191
192    /// Remove a durable pin from a blob.
193    ///
194    /// # Errors
195    ///
196    /// Returns an error if the tag cannot be removed.
197    pub async fn unpin(&self, name: impl AsRef<str>) -> Result<()> {
198        self.store
199            .tags()
200            .delete(name.as_ref())
201            .await
202            .map(|_deleted| ())
203            .map_err(|error| SyncwebError::operation("failed to unpin blob", error))
204    }
205
206    /// # Errors
207    ///
208    /// Returns an error if the pin cannot be queried.
209    pub async fn is_pinned(&self, name: impl AsRef<str>, hash: Hash) -> Result<bool> {
210        self.store
211            .tags()
212            .get(name.as_ref())
213            .await
214            .map(|tag_info| tag_info.is_some_and(|tag| tag.hash == hash))
215            .map_err(|error| SyncwebError::operation("failed to query blob pin", error))
216    }
217
218    /// List all complete blobs currently in the store.
219    ///
220    /// # Errors
221    ///
222    /// Returns an error if the store cannot be queried.
223    pub async fn list_hashes(&self) -> Result<Vec<Hash>> {
224        self.store
225            .blobs()
226            .list()
227            .hashes()
228            .await
229            .map_err(|error| SyncwebError::operation("failed to list blobs", error))
230    }
231
232    /// List named pins whose names start with `prefix`.
233    ///
234    /// # Errors
235    ///
236    /// Returns an error if the store cannot be queried.
237    pub async fn list_pins(&self, prefix: impl AsRef<[u8]>) -> Result<Vec<(String, Hash)>> {
238        let mut tags = self
239            .store
240            .tags()
241            .list_prefix(prefix)
242            .await
243            .map_err(|error| SyncwebError::operation("failed to list blob pins", error))?;
244        let mut pins = Vec::new();
245        while let Some(tag_result) = tags.next().await {
246            let tag = tag_result.map_err(|error| SyncwebError::operation("failed to read blob pin", error))?;
247            let name = String::from_utf8(tag.name.as_ref().to_vec())
248                .map_err(|error| SyncwebError::operation("blob pin name is not UTF-8", error))?;
249            pins.push((name, tag.hash));
250        }
251        Ok(pins)
252    }
253
254    /// # Errors
255    ///
256    /// Returns an error if the blob cannot be fetched from the remote endpoint.
257    pub async fn fetch(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<()> {
258        self.address_lookup.add_endpoint_info(ticket.addr().clone());
259        self.store
260            .downloader(endpoint)
261            .download(ticket.hash_and_format(), [ticket.addr().id])
262            .await
263            .map_err(|error| SyncwebError::operation("failed to fetch blob", error))?;
264        Ok(())
265    }
266
267    /// Fetch a blob with network access control gating.
268    ///
269    /// Checks via `NetworkContext` that the local node can access the blob
270    /// through at least one network membership before downloading.
271    ///
272    /// # Errors
273    ///
274    /// Returns `AccessDenied` if the blob is not associated with any network
275    /// the local node belongs to. Returns other errors if the fetch itself fails.
276    pub async fn download_with_network(
277        &self,
278        endpoint: &Endpoint,
279        ticket: &BlobTicket,
280        ctx: &NetworkContext,
281    ) -> Result<()> {
282        let hash = ticket.hash();
283        if !ctx.can_access_blob(&hash)? {
284            return Err(SyncwebError::access_denied("blob not accessible in any network"));
285        }
286        self.fetch(endpoint, ticket).await
287    }
288
289    /// Force-fetch a blob from a remote peer, re-downloading even if the blob
290    /// is already present locally. Uses `iroh-blobs`' low-level `execute_get`
291    /// API which does not check local availability.
292    ///
293    /// # Errors
294    ///
295    /// Returns an error if the connection or download fails.
296    pub async fn force_fetch(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<()> {
297        self.address_lookup.add_endpoint_info(ticket.addr().clone());
298        let hash = ticket.hash();
299        let conn = endpoint
300            .connect(ticket.addr().id, iroh_blobs::protocol::ALPN)
301            .await
302            .map_err(|error| SyncwebError::operation("failed to connect to provider", error))?;
303        self.force_fetch_from_conn(conn, hash).await
304    }
305
306    /// Force-fetch a blob from a known peer, connecting by public key alone.
307    ///
308    /// # Errors
309    ///
310    /// Returns an error if the connection or download fails.
311    pub async fn force_fetch_from_peer(&self, endpoint: &Endpoint, peer: &iroh::PublicKey, hash: Hash) -> Result<()> {
312        let conn = endpoint
313            .connect(*peer, iroh_blobs::protocol::ALPN)
314            .await
315            .map_err(|error| SyncwebError::operation("failed to connect to peer", error))?;
316        self.force_fetch_from_conn(conn, hash).await
317    }
318
319    async fn force_fetch_from_conn(&self, conn: iroh::endpoint::Connection, hash: Hash) -> Result<()> {
320        let request = GetRequest::blob(hash);
321        self.store
322            .remote()
323            .execute_get(conn, request)
324            .await
325            .map_err(|error| SyncwebError::operation("failed to force-fetch blob", error))?;
326        Ok(())
327    }
328}
syncweb_core/node/blob_store.rs

syncweb_core/node/
blob_store.rs
