syncweb_core/indexing/
catalog.rs

1//! Public metadata catalogs for the opt-in indexing service.
2//!
3//! Catalogs are ordinary iroh-docs namespaces. The catalog contains metadata
4//! only; the synchronized folder remains the source of the actual content.
5
6use std::{
7    collections::HashMap,
8    path::Path,
9    sync::{Arc, Mutex},
10};
11
12use iroh_blobs::Hash;
13use iroh_docs::{
14    AuthorId, DocTicket, Entry, NamespaceId,
15    api::{Doc, protocol::ShareMode},
16    engine::LiveEvent,
17};
18use n0_future::StreamExt;
19use serde::{Deserialize, Serialize};
20use tokio::task::JoinHandle;
21
22use crate::{
23    error::{Result, SyncwebError},
24    folder::SyncwebFolder,
25    indexing::{IndexingDatabase, IndexingService},
26    node::{blob_store::BlobStore, docs_engine::DocsEngine},
27};
28
29const CATALOG_METADATA_KEY: &[u8] = b"sys/syncweb/catalog/metadata";
30const CATALOG_RECORD_PREFIX: &[u8] = b"record/";
31const RETRY_COUNT: usize = 20;
32
33/// Metadata describing a catalog namespace.
34#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
35#[non_exhaustive]
36pub struct CatalogMetadata {
37    pub name: String,
38    pub description: Option<String>,
39    pub publisher: String,
40}
41
42impl CatalogMetadata {
43    /// Create catalog metadata with no description.
44    #[must_use]
45    pub fn new(name: impl Into<String>, publisher: impl Into<String>) -> Self {
46        Self {
47            name: name.into(),
48            description: None,
49            publisher: publisher.into(),
50        }
51    }
52
53    /// # Errors
54    ///
55    /// Returns an error if the catalog name is empty.
56    pub fn validate(&self) -> Result<()> {
57        if self.name.trim().is_empty() {
58            return Err(SyncwebError::InvalidConfig("catalog name cannot be empty".to_owned()));
59        }
60        Ok(())
61    }
62
63    /// # Errors
64    ///
65    /// Returns an error if the metadata cannot be encoded.
66    pub fn to_bytes(&self) -> Result<Vec<u8>> {
67        self.validate()?;
68        serde_json::to_vec(self).map_err(|error| SyncwebError::operation("failed to serialize catalog metadata", error))
69    }
70
71    /// # Errors
72    ///
73    /// Returns an error if the metadata cannot be decoded or is invalid.
74    pub fn from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self> {
75        let metadata: Self = serde_json::from_slice(bytes.as_ref())
76            .map_err(|error| SyncwebError::operation("failed to deserialize catalog metadata", error))?;
77        metadata.validate()?;
78        Ok(metadata)
79    }
80}
81
82/// A searchable metadata record published by a folder.
83#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
84#[non_exhaustive]
85pub struct CatalogRecord {
86    /// The catalog containing this record. This is set by the catalog service
87    /// when a record is written or imported.
88    #[serde(default = "empty_namespace")]
89    pub catalog_namespace_id: NamespaceId,
90    pub folder_namespace_id: NamespaceId,
91    pub key: Vec<u8>,
92    pub hash: Hash,
93    pub size: u64,
94    pub folder_name: String,
95    pub title: String,
96    pub tags: Vec<String>,
97    pub publisher: String,
98}
99
100impl CatalogRecord {
101    /// Create a record using the key's file name as its title.
102    #[must_use]
103    pub fn new(folder_namespace_id: NamespaceId, key: impl AsRef<[u8]>, hash: Hash, size: u64) -> Self {
104        let key_bytes = key.as_ref().to_vec();
105        let key_text = String::from_utf8_lossy(&key_bytes);
106        let title = Path::new(key_text.as_ref())
107            .file_name()
108            .and_then(|name| name.to_str())
109            .unwrap_or_else(|| key_text.as_ref())
110            .to_owned();
111        Self {
112            catalog_namespace_id: empty_namespace(),
113            folder_namespace_id,
114            key: key_bytes,
115            hash,
116            size,
117            folder_name: folder_namespace_id.to_string(),
118            title,
119            tags: Vec::new(),
120            publisher: String::new(),
121        }
122    }
123
124    /// Set the namespace containing this record.
125    #[must_use]
126    pub const fn in_catalog(mut self, namespace_id: NamespaceId) -> Self {
127        self.catalog_namespace_id = namespace_id;
128        self
129    }
130
131    /// # Errors
132    ///
133    /// Returns an error if required searchable metadata is empty.
134    pub fn validate(&self) -> Result<()> {
135        if self.folder_name.trim().is_empty() {
136            return Err(SyncwebError::InvalidConfig(
137                "catalog record folder name cannot be empty".to_owned(),
138            ));
139        }
140        if self.title.trim().is_empty() {
141            return Err(SyncwebError::InvalidConfig(
142                "catalog record title cannot be empty".to_owned(),
143            ));
144        }
145        Ok(())
146    }
147
148    /// # Errors
149    ///
150    /// Returns an error if the record is invalid or cannot be encoded.
151    pub fn to_bytes(&self) -> Result<Vec<u8>> {
152        self.validate()?;
153        serde_json::to_vec(self).map_err(|error| SyncwebError::operation("failed to serialize catalog record", error))
154    }
155
156    /// # Errors
157    ///
158    /// Returns an error if the record cannot be decoded or is invalid.
159    pub fn from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self> {
160        let record: Self = serde_json::from_slice(bytes.as_ref())
161            .map_err(|error| SyncwebError::operation("failed to deserialize catalog record", error))?;
162        record.validate()?;
163        Ok(record)
164    }
165}
166
167/// A local handle to an iroh-docs catalog namespace.
168#[derive(Clone)]
169pub struct Catalog {
170    doc: Doc,
171    namespace_id: NamespaceId,
172    name: String,
173}
174
175impl std::fmt::Debug for Catalog {
176    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
177        formatter
178            .debug_struct("Catalog")
179            .field("namespace_id", &self.namespace_id)
180            .field("name", &self.name)
181            .finish_non_exhaustive()
182    }
183}
184
185impl Catalog {
186    #[must_use]
187    pub fn new(doc: Doc, name: impl Into<String>) -> Self {
188        let namespace_id = doc.id();
189        Self {
190            doc,
191            namespace_id,
192            name: name.into(),
193        }
194    }
195
196    #[must_use]
197    pub const fn doc(&self) -> &Doc {
198        &self.doc
199    }
200
201    #[must_use]
202    pub const fn namespace_id(&self) -> NamespaceId {
203        self.namespace_id
204    }
205
206    #[must_use]
207    pub fn name(&self) -> &str {
208        &self.name
209    }
210}
211
212/// The catalog publisher and subscriber for an indexing database.
213#[derive(Clone)]
214pub struct CatalogService {
215    indexing: IndexingService,
216    docs: DocsEngine,
217    blobs: BlobStore,
218    author: AuthorId,
219    tasks: Arc<Mutex<HashMap<NamespaceId, JoinHandle<()>>>>,
220}
221
222impl std::fmt::Debug for CatalogService {
223    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
224        formatter.debug_struct("CatalogService").finish_non_exhaustive()
225    }
226}
227
228impl CatalogService {
229    /// Construct a catalog service using an existing indexing database.
230    #[must_use]
231    pub fn new(indexing: &IndexingService, docs: &DocsEngine, blobs: &BlobStore, author: AuthorId) -> Self {
232        Self {
233            indexing: indexing.clone(),
234            docs: docs.clone(),
235            blobs: blobs.clone(),
236            author,
237            tasks: Arc::new(Mutex::new(HashMap::new())),
238        }
239    }
240
241    #[must_use]
242    pub const fn database(&self) -> &IndexingDatabase {
243        self.indexing.database()
244    }
245
246    #[must_use]
247    pub const fn indexing(&self) -> &IndexingService {
248        &self.indexing
249    }
250
251    /// Create and register a catalog namespace.
252    ///
253    /// # Errors
254    ///
255    /// Returns an error if the namespace or its metadata cannot be created.
256    pub async fn create_catalog(&self, name: impl Into<String>) -> Result<Catalog> {
257        let metadata = CatalogMetadata::new(name, self.author.to_string());
258        metadata.validate()?;
259        let doc = self.docs.create_namespace().await?;
260        let catalog = Catalog::new(doc, metadata.name.clone());
261        self.docs
262            .set(&catalog.doc, self.author, CATALOG_METADATA_KEY, metadata.to_bytes()?)
263            .await?;
264        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
265        Ok(catalog)
266    }
267
268    /// Create a catalog namespace. Alias for [`Self::create_catalog`].
269    ///
270    /// # Errors
271    ///
272    /// Returns an error if the namespace or its metadata cannot be created.
273    pub async fn create(&self, name: impl Into<String>) -> Result<Catalog> {
274        self.create_catalog(name).await
275    }
276
277    /// Publish all entries in a folder to a catalog.
278    ///
279    /// The catalog contains searchable metadata and content hashes, not file
280    /// contents. The folder's existing document remains authoritative.
281    ///
282    /// # Errors
283    ///
284    /// Returns an error if folder entries or catalog records cannot be read or
285    /// written.
286    pub async fn publish_folder(&self, catalog: &Catalog, folder: &SyncwebFolder) -> Result<usize> {
287        self.publish_folder_with_metadata(catalog, folder, folder.namespace_id().to_string(), &[])
288            .await
289    }
290
291    /// Publish all entries in a folder with a display name and search tags.
292    ///
293    /// # Errors
294    ///
295    /// Returns an error if folder entries or catalog records cannot be read or
296    /// written.
297    pub async fn publish_folder_with_metadata(
298        &self,
299        catalog: &Catalog,
300        folder: &SyncwebFolder,
301        folder_name: impl Into<String>,
302        tags: &[String],
303    ) -> Result<usize> {
304        let folder_name_value = folder_name.into();
305        if folder_name_value.trim().is_empty() {
306            return Err(SyncwebError::InvalidConfig(
307                "catalog folder name cannot be empty".to_owned(),
308            ));
309        }
310        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
311        let entries = folder.docs_engine().list_latest(folder.doc()).await?;
312        let mut published = 0_usize;
313        for entry in entries {
314            if is_system_key(entry.key()) {
315                continue;
316            }
317            let mut record = CatalogRecord::new(
318                folder.namespace_id(),
319                entry.key(),
320                entry.content_hash(),
321                entry.content_len(),
322            )
323            .in_catalog(catalog.namespace_id);
324            record.folder_name.clone_from(&folder_name_value);
325            record.tags = tags.to_vec();
326            record.publisher = self.author.to_string();
327            let key = record_key(&record);
328            self.docs
329                .set(&catalog.doc, self.author, key, record.to_bytes()?)
330                .await?;
331            self.database().upsert_catalog_record(&record)?;
332            published = published.saturating_add(1);
333        }
334        Ok(published)
335    }
336
337    /// Publish a folder using its namespace as the display name.
338    ///
339    /// # Errors
340    ///
341    /// Returns an error if folder entries or catalog records cannot be read or
342    /// written.
343    pub async fn publish(&self, catalog: &Catalog, folder: &SyncwebFolder) -> Result<usize> {
344        self.publish_folder(catalog, folder).await
345    }
346
347    /// Create a read-only ticket for a catalog namespace.
348    ///
349    /// # Errors
350    ///
351    /// Returns an error if the catalog ticket cannot be created.
352    pub async fn ticket(&self, catalog: &Catalog, endpoint: iroh::EndpointAddr, writable: bool) -> Result<DocTicket> {
353        let mode = if writable { ShareMode::Write } else { ShareMode::Read };
354        self.docs.share(&catalog.doc, mode, endpoint).await
355    }
356
357    /// Subscribe to a catalog using its iroh-docs ticket.
358    ///
359    /// Subscription starts document synchronization and keeps a background
360    /// watcher which imports new records into the local FTS database.
361    ///
362    /// # Errors
363    ///
364    /// Returns an error if the ticket cannot be imported or synchronized.
365    pub async fn subscribe(&self, ticket: DocTicket) -> Result<Catalog> {
366        let doc = self.docs.import_ticket(ticket).await?;
367        let catalog = Catalog::new(doc, "remote catalog");
368        self.activate(&catalog).await?;
369        Ok(catalog)
370    }
371
372    /// Subscribe to a catalog. Alias for [`Self::subscribe`].
373    ///
374    /// # Errors
375    ///
376    /// Returns an error if the ticket cannot be imported or synchronized.
377    pub async fn subscribe_catalog(&self, ticket: DocTicket) -> Result<Catalog> {
378        self.subscribe(ticket).await
379    }
380
381    /// Open and subscribe to a catalog namespace already available locally.
382    ///
383    /// # Errors
384    ///
385    /// Returns an error if the namespace is not available or cannot be synced.
386    pub async fn subscribe_namespace(&self, namespace_id: NamespaceId) -> Result<Catalog> {
387        let doc = self
388            .docs
389            .open(namespace_id)
390            .await?
391            .ok_or(SyncwebError::NamespaceNotAvailable)?;
392        let catalog = Catalog::new(doc, "remote catalog");
393        self.activate(&catalog).await?;
394        Ok(catalog)
395    }
396
397    /// Synchronize a catalog once and import all records currently available.
398    ///
399    /// # Errors
400    ///
401    /// Returns an error if document synchronization or record indexing fails.
402    pub async fn sync_catalog(&self, catalog: &Catalog) -> Result<usize> {
403        self.docs.start_sync(&catalog.doc, Vec::new()).await?;
404        self.index_catalog(catalog).await
405    }
406
407    /// Synchronize a catalog once. Alias for [`Self::sync_catalog`].
408    ///
409    /// # Errors
410    ///
411    /// Returns an error if document synchronization or record indexing fails.
412    pub async fn sync(&self, catalog: &Catalog) -> Result<usize> {
413        self.sync_catalog(catalog).await
414    }
415
416    /// Import all currently available records from a catalog.
417    ///
418    /// # Errors
419    ///
420    /// Returns an error if catalog entries cannot be read or decoded.
421    pub async fn index_catalog(&self, catalog: &Catalog) -> Result<usize> {
422        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
423        let entries = self.docs.list_latest(&catalog.doc).await?;
424        let mut indexed = 0_usize;
425        for entry in entries {
426            if self.index_catalog_entry(catalog.namespace_id, &entry).await? {
427                indexed = indexed.saturating_add(1);
428            }
429        }
430        Ok(indexed)
431    }
432
433    /// Search records imported from all known catalogs.
434    ///
435    /// # Errors
436    ///
437    /// Returns an error if the FTS query is invalid.
438    pub fn search(&self, query: &str, limit: usize) -> Result<Vec<CatalogRecord>> {
439        self.database().search_catalogs(query, limit)
440    }
441
442    /// Search records imported from all known catalogs. Alias for [`Self::search`].
443    ///
444    /// # Errors
445    ///
446    /// Returns an error if the FTS query is invalid.
447    pub fn search_fts(&self, query: &str, limit: usize) -> Result<Vec<CatalogRecord>> {
448        self.search(query, limit)
449    }
450
451    /// Stop following a catalog and remove its local records.
452    ///
453    /// # Errors
454    ///
455    /// Returns an error if the catalog index cannot be removed.
456    pub fn unsubscribe(&self, namespace_id: NamespaceId) -> Result<()> {
457        let task_handle = self
458            .tasks
459            .lock()
460            .map_err(|error| SyncwebError::operation("catalog task lock poisoned", error))?
461            .remove(&namespace_id);
462        if let Some(handle) = task_handle {
463            handle.abort();
464        }
465        self.database().disable_catalog(namespace_id)
466    }
467
468    async fn activate(&self, catalog: &Catalog) -> Result<()> {
469        let live_events = self.docs.watch(&catalog.doc).await?;
470        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
471        self.index_catalog(catalog).await?;
472
473        let namespace_id = catalog.namespace_id;
474        let database = self.indexing.database().clone();
475        let docs = self.docs.clone();
476        let blobs = self.blobs.clone();
477        let task = tokio::spawn(async move {
478            consume_catalog_events(namespace_id, live_events, database, blobs, docs).await;
479        });
480        self.tasks
481            .lock()
482            .map_err(|error| SyncwebError::operation("catalog task lock poisoned", error))?
483            .insert(namespace_id, task);
484        if let Err(error) = self.docs.start_sync(&catalog.doc, Vec::new()).await {
485            if let Ok(mut tasks) = self.tasks.lock()
486                && let Some(task_handle) = tasks.remove(&namespace_id)
487            {
488                task_handle.abort();
489            }
490            return Err(error);
491        }
492        Ok(())
493    }
494
495    async fn index_catalog_entry(&self, namespace_id: NamespaceId, entry: &Entry) -> Result<bool> {
496        if !is_record_key(entry.key()) {
497            return Ok(false);
498        }
499        let bytes = self.blobs.get(entry.content_hash()).await?;
500        let mut record = CatalogRecord::from_bytes(bytes)?;
501        record.catalog_namespace_id = namespace_id;
502        self.database().upsert_catalog_record(&record)?;
503        Ok(true)
504    }
505}
506
507async fn consume_catalog_events(
508    namespace_id: NamespaceId,
509    mut live_events: impl n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin + 'static,
510    database: IndexingDatabase,
511    blobs: BlobStore,
512    docs: DocsEngine,
513) {
514    while let Some(event_result) = live_events.next().await {
515        let event = match event_result {
516            Ok(event) => event,
517            Err(error) => {
518                tracing::error!(%namespace_id, error = %error, "catalog document event failed");
519                break;
520            }
521        };
522        if let Err(error) = process_catalog_event(namespace_id, event, &database, &blobs, &docs).await {
523            tracing::error!(%namespace_id, error = %error, "failed to process catalog event");
524        }
525    }
526}
527
528async fn process_catalog_event(
529    namespace_id: NamespaceId,
530    event: LiveEvent,
531    database: &IndexingDatabase,
532    blobs: &BlobStore,
533    docs: &DocsEngine,
534) -> Result<()> {
535    match event {
536        LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
537            if !is_record_key(entry.key()) {
538                return Ok(());
539            }
540            let mut record = read_catalog_record(blobs, &entry).await?;
541            record.catalog_namespace_id = namespace_id;
542            database.upsert_catalog_record(&record)?;
543        }
544        LiveEvent::ContentReady { .. } => reindex_catalog(docs, blobs, database, namespace_id).await?,
545        LiveEvent::PendingContentReady
546        | LiveEvent::NeighborUp(_)
547        | LiveEvent::NeighborDown(_)
548        | LiveEvent::SyncFinished(_) => {}
549    }
550    Ok(())
551}
552
553async fn reindex_catalog(
554    docs: &DocsEngine,
555    blobs: &BlobStore,
556    database: &IndexingDatabase,
557    namespace_id: NamespaceId,
558) -> Result<()> {
559    let doc = docs
560        .open(namespace_id)
561        .await?
562        .ok_or(SyncwebError::NamespaceNotAvailable)?;
563    for entry in docs.list_latest(&doc).await? {
564        if !is_record_key(entry.key()) {
565            continue;
566        }
567        let bytes = blobs.get(entry.content_hash()).await?;
568        let mut record = CatalogRecord::from_bytes(bytes)?;
569        record.catalog_namespace_id = namespace_id;
570        database.upsert_catalog_record(&record)?;
571    }
572    Ok(())
573}
574
575async fn read_catalog_record(blobs: &BlobStore, entry: &Entry) -> Result<CatalogRecord> {
576    let mut last_error = None;
577    for attempt in 0..RETRY_COUNT {
578        match blobs.get(entry.content_hash()).await {
579            Ok(bytes) => return CatalogRecord::from_bytes(bytes),
580            Err(error) => {
581                last_error = Some(error);
582                if attempt.saturating_add(1) < RETRY_COUNT {
583                    tokio::time::sleep(std::time::Duration::from_millis(25)).await;
584                }
585            }
586        }
587    }
588    Err(last_error.unwrap_or_else(|| SyncwebError::InvalidConfig("catalog record content is unavailable".to_owned())))
589}
590
591fn record_key(record: &CatalogRecord) -> Vec<u8> {
592    format!(
593        "{}{}/{}",
594        String::from_utf8_lossy(CATALOG_RECORD_PREFIX),
595        record.folder_namespace_id,
596        hex::encode(&record.key)
597    )
598    .into_bytes()
599}
600
601fn is_record_key(key: &[u8]) -> bool {
602    key.starts_with(CATALOG_RECORD_PREFIX)
603}
604
605fn is_system_key(key: &[u8]) -> bool {
606    key.starts_with(b"sys/")
607}
608
609fn empty_namespace() -> NamespaceId {
610    NamespaceId::from([0_u8; 32])
611}
syncweb_core/indexing/catalog.rs

syncweb_core/indexing/
catalog.rs
