syncweb-core 0.1.1-alpha.7

//! Public metadata catalogs for the opt-in indexing service.
//!
//! Catalogs are iroh-docs namespaces provisioned exactly like folders (via
//! [`crate::node::docs_engine::DocsEngine::provision`]): a catalog is a folder
//! whose entries are [`CatalogRecord`] metadata, plus an
//! [`IndexingDatabase::enable_catalog`] registration. The catalog contains
//! metadata only; the synchronized folder remains the source of the actual
//! content.

use std::{
    collections::HashMap,
    path::Path,
    sync::{Arc, Mutex},
};

use iroh_blobs::Hash;
use iroh_docs::{AuthorId, DocTicket, Entry, NamespaceId, api::Doc, engine::LiveEvent};
use n0_future::StreamExt;
use serde::{Deserialize, Serialize};
use tokio::task::JoinHandle;

use crate::{
    error::{Result, SyncwebError},
    folder::SyncwebFolder,
    indexing::{IndexingDatabase, IndexingService},
    node::{blob_store::BlobStore, docs_engine::DocsEngine},
};
const CATALOG_RECORD_PREFIX: &[u8] = b"record/";
const RETRY_COUNT: usize = 20;
const MAX_CONSECUTIVE_STREAM_ERRORS: usize = 5;
const STREAM_RETRY_BASE_DELAY: std::time::Duration = std::time::Duration::from_secs(1);
const STREAM_RETRY_MAX_DELAY: std::time::Duration = std::time::Duration::from_mins(1);

/// Metadata describing a catalog namespace.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct CatalogMetadata {
    pub name: String,
    pub description: Option<String>,
    pub publisher: String,
}

impl CatalogMetadata {
    /// Create catalog metadata with no description.
    #[must_use]
    pub fn new(name: impl Into<String>, publisher: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            description: None,
            publisher: publisher.into(),
        }
    }

    /// # Errors
    ///
    /// Returns an error if the catalog name is empty.
    pub fn validate(&self) -> Result<()> {
        if self.name.trim().is_empty() {
            return Err(SyncwebError::InvalidConfig("catalog name cannot be empty".to_owned()));
        }
        Ok(())
    }

    /// # Errors
    ///
    /// Returns an error if the metadata cannot be encoded.
    pub fn to_bytes(&self) -> Result<Vec<u8>> {
        self.validate()?;
        serde_json::to_vec(self).map_err(|error| SyncwebError::operation("failed to serialize catalog metadata", error))
    }

    /// # Errors
    ///
    /// Returns an error if the metadata cannot be decoded or is invalid.
    pub fn from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self> {
        let metadata: Self = serde_json::from_slice(bytes.as_ref())
            .map_err(|error| SyncwebError::operation("failed to deserialize catalog metadata", error))?;
        metadata.validate()?;
        Ok(metadata)
    }
}

/// A searchable metadata record published by a folder.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct CatalogRecord {
    /// The catalog containing this record. This is set by the catalog service
    /// when a record is written or imported.
    #[serde(default = "empty_namespace")]
    pub catalog_namespace_id: NamespaceId,
    pub folder_namespace_id: NamespaceId,
    pub key: Vec<u8>,
    pub hash: Hash,
    pub size: u64,
    pub folder_name: String,
    pub title: String,
    pub tags: Vec<String>,
    pub publisher: String,
}

impl CatalogRecord {
    /// Create a record using the key's file name as its title.
    #[must_use]
    pub fn new(folder_namespace_id: NamespaceId, key: impl AsRef<[u8]>, hash: Hash, size: u64) -> Self {
        let key_bytes = key.as_ref().to_vec();
        let key_text = String::from_utf8_lossy(&key_bytes);
        let title = Path::new(key_text.as_ref())
            .file_name()
            .and_then(|name| name.to_str())
            .unwrap_or_else(|| key_text.as_ref())
            .to_owned();
        Self {
            catalog_namespace_id: empty_namespace(),
            folder_namespace_id,
            key: key_bytes,
            hash,
            size,
            folder_name: folder_namespace_id.to_string(),
            title,
            tags: Vec::new(),
            publisher: String::new(),
        }
    }

    /// Set the namespace containing this record.
    #[must_use]
    pub const fn in_catalog(mut self, namespace_id: NamespaceId) -> Self {
        self.catalog_namespace_id = namespace_id;
        self
    }

    /// # Errors
    ///
    /// Returns an error if required searchable metadata is empty.
    pub fn validate(&self) -> Result<()> {
        if self.folder_name.trim().is_empty() {
            return Err(SyncwebError::InvalidConfig(
                "catalog record folder name cannot be empty".to_owned(),
            ));
        }
        if self.title.trim().is_empty() {
            return Err(SyncwebError::InvalidConfig(
                "catalog record title cannot be empty".to_owned(),
            ));
        }
        Ok(())
    }

    /// # Errors
    ///
    /// Returns an error if the record is invalid or cannot be encoded.
    pub fn to_bytes(&self) -> Result<Vec<u8>> {
        self.validate()?;
        serde_json::to_vec(self).map_err(|error| SyncwebError::operation("failed to serialize catalog record", error))
    }

    /// # Errors
    ///
    /// Returns an error if the record cannot be decoded or is invalid.
    pub fn from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self> {
        let record: Self = serde_json::from_slice(bytes.as_ref())
            .map_err(|error| SyncwebError::operation("failed to deserialize catalog record", error))?;
        record.validate()?;
        Ok(record)
    }
}

/// A local handle to an iroh-docs catalog namespace.
#[derive(Clone)]
pub struct Catalog {
    doc: Doc,
    namespace_id: NamespaceId,
    name: String,
}

impl std::fmt::Debug for Catalog {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("Catalog")
            .field("namespace_id", &self.namespace_id)
            .field("name", &self.name)
            .finish_non_exhaustive()
    }
}

impl Catalog {
    #[must_use]
    pub fn new(doc: Doc, name: impl Into<String>) -> Self {
        let namespace_id = doc.id();
        Self {
            doc,
            namespace_id,
            name: name.into(),
        }
    }

    #[must_use]
    pub const fn doc(&self) -> &Doc {
        &self.doc
    }

    #[must_use]
    pub const fn namespace_id(&self) -> NamespaceId {
        self.namespace_id
    }

    #[must_use]
    pub fn name(&self) -> &str {
        &self.name
    }
}

/// The catalog publisher and subscriber for an indexing database.
#[derive(Clone)]
pub struct CatalogService {
    indexing: IndexingService,
    docs: DocsEngine,
    blobs: BlobStore,
    author: AuthorId,
    tasks: Arc<Mutex<HashMap<NamespaceId, JoinHandle<()>>>>,
}

impl std::fmt::Debug for CatalogService {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.debug_struct("CatalogService").finish_non_exhaustive()
    }
}

impl CatalogService {
    /// Construct a catalog service using an existing indexing database.
    #[must_use]
    pub fn new(indexing: &IndexingService, docs: &DocsEngine, blobs: &BlobStore, author: AuthorId) -> Self {
        Self {
            indexing: indexing.clone(),
            docs: docs.clone(),
            blobs: blobs.clone(),
            author,
            tasks: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    #[must_use]
    pub const fn database(&self) -> &IndexingDatabase {
        self.indexing.database()
    }

    #[must_use]
    pub const fn indexing(&self) -> &IndexingService {
        &self.indexing
    }

    /// Create and register a catalog namespace.
    ///
    /// Catalogs are folders that store indexable [`CatalogRecord`] entries:
    /// provisioning is unified with folder creation through
    /// [`DocsEngine::provision`], and the catalog metadata lives under the
    /// `sys/` key namespace like folder metadata.
    ///
    /// # Errors
    ///
    /// Returns an error if the namespace or its metadata cannot be created.
    pub async fn create_catalog(&self, name: impl Into<String>) -> Result<Catalog> {
        let metadata = CatalogMetadata::new(name, self.author.to_string());
        metadata.validate()?;
        let provisioned = self
            .docs
            .provision(
                crate::node::docs_engine::ProvisionKind::Catalog,
                None,
                metadata.to_bytes()?,
            )
            .await?;
        let catalog = Catalog::new(provisioned.doc, metadata.name.clone());
        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
        Ok(catalog)
    }

    /// Return the registered catalog with `name`, creating it when needed.
    ///
    /// An existing registered catalog is opened and subscribed to; otherwise a
    /// fresh catalog namespace is provisioned and registered.
    ///
    /// # Errors
    ///
    /// Returns an error if the namespace cannot be created or opened.
    pub async fn get_or_create_catalog(&self, name: &str) -> Result<Catalog> {
        let registered = self.database().load_catalogs()?;
        if let Some((namespace_id, _)) = registered.into_iter().find(|(_, label)| label == name) {
            return self.subscribe_namespace(namespace_id).await;
        }
        self.create_catalog(name).await
    }

    /// Publish all entries in a folder to a catalog.
    ///
    /// The catalog contains searchable metadata and content hashes, not file
    /// contents. The folder's existing document remains authoritative.
    ///
    /// # Errors
    ///
    /// Returns an error if folder entries or catalog records cannot be read or
    /// written.
    pub async fn publish_folder_with_metadata(
        &self,
        catalog: &Catalog,
        folder: &SyncwebFolder,
        folder_name: impl Into<String>,
        tags: &[String],
    ) -> Result<usize> {
        let folder_name_value = folder_name.into();
        if folder_name_value.trim().is_empty() {
            return Err(SyncwebError::InvalidConfig(
                "catalog folder name cannot be empty".to_owned(),
            ));
        }
        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
        let entries = folder.docs_engine().list_latest(folder.doc()).await?;
        let mut published = 0_usize;
        for entry in entries {
            if is_system_key(entry.key()) {
                continue;
            }
            let mut record = CatalogRecord::new(
                folder.namespace_id(),
                entry.key(),
                entry.content_hash(),
                entry.content_len(),
            )
            .in_catalog(catalog.namespace_id);
            record.folder_name.clone_from(&folder_name_value);
            record.tags = tags.to_vec();
            record.publisher = self.author.to_string();
            let key = record_key(&record);
            self.docs
                .set(&catalog.doc, self.author, key, record.to_bytes()?)
                .await?;
            self.database().upsert_catalog_record(&record)?;
            published = published.saturating_add(1);
        }
        Ok(published)
    }

    /// Create a read-only ticket for a catalog namespace.
    ///
    /// # Errors
    ///
    /// Returns an error if the catalog ticket cannot be created.
    pub async fn ticket(&self, catalog: &Catalog, writable: bool) -> Result<DocTicket> {
        self.docs.share_ticket(&catalog.doc, writable).await
    }

    /// Subscribe to a catalog using its iroh-docs ticket.
    ///
    /// Subscription starts document synchronization and keeps a background
    /// watcher which imports new records into the local FTS database.
    ///
    /// # Errors
    ///
    /// Returns an error if the ticket cannot be imported or synchronized.
    pub async fn subscribe(&self, ticket: DocTicket) -> Result<Catalog> {
        let doc = self.docs.import_ticket(ticket).await?;
        let catalog = Catalog::new(doc, "remote catalog");
        self.activate(&catalog).await?;
        Ok(catalog)
    }

    /// Open and subscribe to a catalog namespace already available locally.
    ///
    /// # Errors
    ///
    /// Returns an error if the namespace is not available or cannot be synced.
    pub async fn subscribe_namespace(&self, namespace_id: NamespaceId) -> Result<Catalog> {
        let (doc, _ticket) = self.docs.create_or_open_namespace(Some(namespace_id)).await?;
        let catalog = Catalog::new(doc, "remote catalog");
        self.activate(&catalog).await?;
        Ok(catalog)
    }

    /// Synchronize a catalog once and import all records currently available.
    ///
    /// # Errors
    ///
    /// Returns an error if document synchronization or record indexing fails.
    pub async fn sync_catalog(&self, catalog: &Catalog) -> Result<usize> {
        self.docs.start_sync(&catalog.doc, Vec::new()).await?;
        self.index_catalog(catalog).await
    }

    /// Import all currently available records from a catalog.
    ///
    /// # Errors
    ///
    /// Returns an error if catalog entries cannot be read or decoded.
    pub async fn index_catalog(&self, catalog: &Catalog) -> Result<usize> {
        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
        let entries = self.docs.list_latest(&catalog.doc).await?;
        let mut indexed = 0_usize;
        for entry in entries {
            if self.index_catalog_entry(catalog.namespace_id, &entry).await? {
                indexed = indexed.saturating_add(1);
            }
        }
        Ok(indexed)
    }

    /// Search records imported from all known catalogs.
    ///
    /// Records blocked by the local denylist are excluded from discovery.
    ///
    /// # Errors
    ///
    /// Returns an error if the FTS query is invalid.
    pub fn search(&self, query: &str, limit: usize) -> Result<Vec<CatalogRecord>> {
        self.indexing.search(query, limit)
    }

    /// Stop following a catalog and remove its local records.
    ///
    /// # Errors
    ///
    /// Returns an error if the catalog index cannot be removed.
    pub fn unsubscribe(&self, namespace_id: NamespaceId) -> Result<()> {
        let task_handle = self
            .tasks
            .lock()
            .map_err(|error| SyncwebError::operation("catalog task lock poisoned", error))?
            .remove(&namespace_id);
        if let Some(handle) = task_handle {
            handle.abort();
        }
        self.database().disable_catalog(namespace_id)
    }

    async fn activate(&self, catalog: &Catalog) -> Result<()> {
        let live_events = self.docs.watch_and_sync(&catalog.doc).await?;
        self.database().enable_catalog(catalog.namespace_id, catalog.name())?;
        self.index_catalog(catalog).await?;

        let namespace_id = catalog.namespace_id;
        let database = self.indexing.database().clone();
        let docs = self.docs.clone();
        let blobs = self.blobs.clone();
        let task = tokio::spawn(async move {
            consume_catalog_events(namespace_id, live_events, database, blobs, docs).await;
        });
        self.tasks
            .lock()
            .map_err(|error| SyncwebError::operation("catalog task lock poisoned", error))?
            .insert(namespace_id, task);
        Ok(())
    }

    async fn index_catalog_entry(&self, namespace_id: NamespaceId, entry: &Entry) -> Result<bool> {
        if !is_record_key(entry.key()) {
            return Ok(false);
        }
        let bytes = self.blobs.get(entry.content_hash()).await?;
        let mut record = CatalogRecord::from_bytes(bytes)?;
        record.catalog_namespace_id = namespace_id;
        self.database().upsert_catalog_record(&record)?;
        Ok(true)
    }
}

async fn consume_catalog_events(
    namespace_id: NamespaceId,
    live_events: impl n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin + 'static,
    database: IndexingDatabase,
    blobs: BlobStore,
    docs: DocsEngine,
) {
    let mut consecutive_errors = 0_usize;
    let mut current_stream: Option<Box<dyn n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin>> =
        Some(Box::new(live_events));
    loop {
        let Some(stream) = current_stream.as_mut() else {
            match reconnect_catalog_stream(namespace_id, &docs, &mut consecutive_errors).await {
                Some(new_stream) => {
                    current_stream = Some(new_stream);
                    continue;
                }
                None if consecutive_errors >= MAX_CONSECUTIVE_STREAM_ERRORS => return,
                None => continue,
            }
        };
        match stream.next().await {
            Some(Ok(event)) => {
                consecutive_errors = 0;
                if let Err(error) = process_catalog_event(namespace_id, event, &database, &blobs, &docs).await {
                    tracing::error!(%namespace_id, error = %error, "failed to process catalog event");
                }
            }
            Some(Err(error)) => {
                tracing::error!(%namespace_id, error = %error, "catalog document event stream error");
                consecutive_errors = consecutive_errors.saturating_add(1);
                current_stream = None;
            }
            None => {
                tracing::info!(%namespace_id, "catalog document event stream ended");
                return;
            }
        }
    }
}

async fn reconnect_catalog_stream(
    namespace_id: NamespaceId,
    docs: &DocsEngine,
    consecutive_errors: &mut usize,
) -> Option<Box<dyn n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin>> {
    if *consecutive_errors >= MAX_CONSECUTIVE_STREAM_ERRORS {
        tracing::error!(
            %namespace_id,
            consecutive_errors = *consecutive_errors,
            "catalog watcher gave up after max retries"
        );
        return None;
    }
    let delay = STREAM_RETRY_BASE_DELAY
        .checked_mul(2_u32.pow(u32::try_from(*consecutive_errors).unwrap_or(0)))
        .unwrap_or(STREAM_RETRY_MAX_DELAY)
        .min(STREAM_RETRY_MAX_DELAY);
    tracing::warn!(%namespace_id, delay_ms = delay.as_millis(), "catalog watcher reconnecting");
    tokio::time::sleep(delay).await;
    let Ok(Some(doc)) = docs.open(namespace_id).await else {
        *consecutive_errors = consecutive_errors.saturating_add(1);
        return None;
    };
    match docs.watch(&doc).await {
        Ok(new_stream) => {
            *consecutive_errors = 0;
            Some(Box::new(new_stream))
        }
        Err(error) => {
            tracing::error!(%namespace_id, error = %error, "catalog watcher re-subscribe failed");
            *consecutive_errors = consecutive_errors.saturating_add(1);
            None
        }
    }
}

async fn process_catalog_event(
    namespace_id: NamespaceId,
    event: LiveEvent,
    database: &IndexingDatabase,
    blobs: &BlobStore,
    docs: &DocsEngine,
) -> Result<()> {
    match event {
        LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
            if !is_record_key(entry.key()) {
                return Ok(());
            }
            let mut record = read_catalog_record(blobs, &entry).await?;
            record.catalog_namespace_id = namespace_id;
            database.upsert_catalog_record(&record)?;
        }
        LiveEvent::ContentReady { hash } => {
            let doc = docs
                .open(namespace_id)
                .await?
                .ok_or(SyncwebError::NamespaceNotAvailable)?;
            for entry in docs.list_latest(&doc).await? {
                if !is_record_key(entry.key()) || entry.content_hash() != hash {
                    continue;
                }
                let Ok(bytes) = blobs.get(entry.content_hash()).await else {
                    continue;
                };
                let mut record = CatalogRecord::from_bytes(bytes)?;
                record.catalog_namespace_id = namespace_id;
                database.upsert_catalog_record(&record)?;
                break;
            }
        }
        LiveEvent::PendingContentReady
        | LiveEvent::NeighborUp(_)
        | LiveEvent::NeighborDown(_)
        | LiveEvent::SyncFinished(_) => {}
    }
    Ok(())
}

async fn read_catalog_record(blobs: &BlobStore, entry: &Entry) -> Result<CatalogRecord> {
    let mut last_error = None;
    for attempt in 0..RETRY_COUNT {
        match blobs.get(entry.content_hash()).await {
            Ok(bytes) => return CatalogRecord::from_bytes(bytes),
            Err(error) => {
                last_error = Some(error);
                if attempt.saturating_add(1) < RETRY_COUNT {
                    tokio::time::sleep(std::time::Duration::from_millis(25)).await;
                }
            }
        }
    }
    Err(last_error.unwrap_or_else(|| SyncwebError::InvalidConfig("catalog record content is unavailable".to_owned())))
}

fn record_key(record: &CatalogRecord) -> Vec<u8> {
    format!(
        "{}{}/{}",
        String::from_utf8_lossy(CATALOG_RECORD_PREFIX),
        record.folder_namespace_id,
        hex::encode(&record.key)
    )
    .into_bytes()
}

fn is_record_key(key: &[u8]) -> bool {
    key.starts_with(CATALOG_RECORD_PREFIX)
}

fn is_system_key(key: &[u8]) -> bool {
    key.starts_with(b"sys/")
}

fn empty_namespace() -> NamespaceId {
    NamespaceId::from([0_u8; 32])
}