syncweb_core/indexing/
catalog.rs

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

syncweb_core/indexing/
catalog.rs
