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, ImportMode},
8    },
9    ticket::BlobTicket,
10};
11use n0_future::StreamExt;
12use std::path::Path;
13
14use crate::error::{Result, SyncwebError};
15
16#[derive(Clone)]
17pub struct BlobStore {
18    store: BlobApi,
19    address_lookup: MemoryLookup,
20}
21
22impl BlobStore {
23    #[must_use]
24    pub fn new(protocol: &BlobsProtocol) -> Self {
25        Self::new_with_address_lookup(protocol, MemoryLookup::new())
26    }
27
28    #[must_use]
29    pub fn new_with_address_lookup(protocol: &BlobsProtocol, address_lookup: MemoryLookup) -> Self {
30        Self {
31            store: protocol.store().clone(),
32            address_lookup,
33        }
34    }
35
36    #[must_use]
37    pub const fn inner(&self) -> &BlobApi {
38        &self.store
39    }
40
41    /// # Errors
42    ///
43    /// Returns an error if the blob fails to be added to the store.
44    pub async fn add_bytes(&self, data: impl AsRef<[u8]>) -> Result<Hash> {
45        Ok(self
46            .store
47            .add_bytes(Bytes::copy_from_slice(data.as_ref()))
48            .await
49            .map_err(|error| SyncwebError::operation("failed to add blob bytes", error))?
50            .hash)
51    }
52
53    /// # Errors
54    ///
55    /// Returns an error if the file fails to be read or added to the store.
56    pub async fn add_file(&self, path: impl AsRef<Path>) -> Result<Hash> {
57        Ok(self
58            .store
59            .add_path(path)
60            .await
61            .map_err(|error| SyncwebError::operation("failed to add blob file", error))?
62            .hash)
63    }
64
65    /// Add a file using a reference to its original path instead of copying
66    /// its contents into the blob store.
67    ///
68    /// The source file must remain available and unchanged for as long as the
69    /// blob store may need to read this blob.
70    ///
71    /// # Errors
72    ///
73    /// Returns an error if the file fails to be read or added to the store.
74    pub async fn add_file_ref(&self, path: impl AsRef<Path>) -> Result<Hash> {
75        Ok(self
76            .store
77            .add_path_with_opts(AddPathOptions {
78                path: path.as_ref().to_owned(),
79                mode: ImportMode::TryReference,
80                format: BlobFormat::Raw,
81            })
82            .await
83            .map_err(|error| SyncwebError::operation("failed to add blob file (reference)", error))?
84            .hash)
85    }
86
87    /// Export a complete blob to a temporary or caller-managed path.
88    ///
89    /// # Errors
90    ///
91    /// Returns an error if the blob cannot be read or written.
92    pub async fn export_to_path(&self, hash: Hash, destination: impl AsRef<Path>) -> Result<u64> {
93        self.store
94            .export(hash, destination)
95            .await
96            .map_err(|error| SyncwebError::operation("failed to export blob", error))
97    }
98
99    /// # Errors
100    ///
101    /// Returns an error if the store cannot be queried.
102    pub async fn has(&self, hash: Hash) -> Result<bool> {
103        self.store
104            .has(hash)
105            .await
106            .map_err(|error| SyncwebError::operation("failed to query blob store", error))
107    }
108
109    /// # Errors
110    ///
111    /// Returns an error if the blob cannot be found or read.
112    pub async fn get(&self, hash: Hash) -> Result<Bytes> {
113        self.store
114            .get_bytes(hash)
115            .await
116            .map_err(|error| SyncwebError::operation("failed to read blob", error))
117    }
118
119    #[must_use]
120    pub fn ticket(&self, endpoint: &Endpoint, hash: Hash) -> BlobTicket {
121        self.ticket_for_addr(endpoint.addr(), hash)
122    }
123
124    #[must_use]
125    pub fn ticket_for_addr(&self, addr: EndpointAddr, hash: Hash) -> BlobTicket {
126        BlobTicket::new(addr, hash, iroh_blobs::BlobFormat::Raw)
127    }
128
129    /// Pin a blob with a durable named tag so garbage collection cannot remove it.
130    ///
131    /// # Errors
132    ///
133    /// Returns an error if the tag cannot be written.
134    pub async fn pin(&self, name: impl AsRef<str>, hash: Hash) -> Result<()> {
135        self.store
136            .tags()
137            .set(name.as_ref(), hash)
138            .await
139            .map_err(|error| SyncwebError::operation("failed to pin blob", error))
140    }
141
142    /// Remove a durable pin from a blob.
143    ///
144    /// # Errors
145    ///
146    /// Returns an error if the tag cannot be removed.
147    pub async fn unpin(&self, name: impl AsRef<str>) -> Result<()> {
148        self.store
149            .tags()
150            .delete(name.as_ref())
151            .await
152            .map(|_deleted| ())
153            .map_err(|error| SyncwebError::operation("failed to unpin blob", error))
154    }
155
156    /// # Errors
157    ///
158    /// Returns an error if the pin cannot be queried.
159    pub async fn is_pinned(&self, name: impl AsRef<str>, hash: Hash) -> Result<bool> {
160        self.store
161            .tags()
162            .get(name.as_ref())
163            .await
164            .map(|tag_info| tag_info.is_some_and(|tag| tag.hash == hash))
165            .map_err(|error| SyncwebError::operation("failed to query blob pin", error))
166    }
167
168    /// List all complete blobs currently in the store.
169    ///
170    /// # Errors
171    ///
172    /// Returns an error if the store cannot be queried.
173    pub async fn list_hashes(&self) -> Result<Vec<Hash>> {
174        self.store
175            .blobs()
176            .list()
177            .hashes()
178            .await
179            .map_err(|error| SyncwebError::operation("failed to list blobs", error))
180    }
181
182    /// List named pins whose names start with `prefix`.
183    ///
184    /// # Errors
185    ///
186    /// Returns an error if the store cannot be queried.
187    pub async fn list_pins(&self, prefix: impl AsRef<[u8]>) -> Result<Vec<(String, Hash)>> {
188        let mut tags = self
189            .store
190            .tags()
191            .list_prefix(prefix)
192            .await
193            .map_err(|error| SyncwebError::operation("failed to list blob pins", error))?;
194        let mut pins = Vec::new();
195        while let Some(tag_result) = tags.next().await {
196            let tag = tag_result.map_err(|error| SyncwebError::operation("failed to read blob pin", error))?;
197            let name = String::from_utf8(tag.name.as_ref().to_vec())
198                .map_err(|error| SyncwebError::operation("blob pin name is not UTF-8", error))?;
199            pins.push((name, tag.hash));
200        }
201        Ok(pins)
202    }
203
204    /// # Errors
205    ///
206    /// Returns an error if the blob cannot be fetched from the remote endpoint.
207    pub async fn fetch(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<()> {
208        self.address_lookup.add_endpoint_info(ticket.addr().clone());
209        self.store
210            .downloader(endpoint)
211            .download(ticket.hash_and_format(), [ticket.addr().id])
212            .await
213            .map_err(|error| SyncwebError::operation("failed to fetch blob", error))?;
214        Ok(())
215    }
216}
syncweb_core/node/blob_store.rs

syncweb_core/node/
blob_store.rs
