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

syncweb_core/indexing/
catalog.rs
