feat: move core services to services directory

src/services/async_item_service.rs

@@ -0,0 +1,116 @@
+use crate::services::error::CoreError;
+use crate::services::item_service::ItemService;
+use crate::services::types::{ItemWithContent, ItemWithMeta};
+use rusqlite::Connection;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+
+/// An asynchronous wrapper around the `ItemService` for use in async contexts like the web server.
+/// It uses `tokio::task::spawn_blocking` to run synchronous database and filesystem operations
+/// on a dedicated thread pool, preventing them from blocking the async runtime.
+#[allow(dead_code)]
+pub struct AsyncItemService {
+    data_path: PathBuf,
+    db: Arc<Mutex<Connection>>,
+}
+
+#[allow(dead_code)]
+impl AsyncItemService {
+    pub fn new(data_path: PathBuf, db: Arc<Mutex<Connection>>) -> Self {
+        Self { data_path, db }
+    }
+
+    pub async fn get_item(&self, id: i64) -> Result<ItemWithMeta, CoreError> {
+        let data_path = self.data_path.clone();
+        let db = self.db.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let conn = db.blocking_lock();
+            let item_service = ItemService::new(data_path);
+            item_service.get_item(&conn, id)
+        })
+        .await
+        .unwrap() // Propagate panics from spawn_blocking
+    }
+
+    pub async fn get_item_content(&self, id: i64) -> Result<ItemWithContent, CoreError> {
+        let data_path = self.data_path.clone();
+        let db = self.db.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let conn = db.blocking_lock();
+            let item_service = ItemService::new(data_path);
+            item_service.get_item_content(&conn, id)
+        })
+        .await
+        .unwrap()
+    }
+
+    pub async fn find_item(
+        &self,
+        ids: Vec<i64>,
+        tags: Vec<String>,
+        meta: HashMap<String, String>,
+    ) -> Result<ItemWithMeta, CoreError> {
+        let data_path = self.data_path.clone();
+        let db = self.db.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let conn = db.blocking_lock();
+            let item_service = ItemService::new(data_path);
+            item_service.find_item(&conn, &ids, &tags, &meta)
+        })
+        .await
+        .unwrap()
+    }
+
+    pub async fn list_items(
+        &self,
+        tags: Vec<String>,
+        meta: HashMap<String, String>,
+    ) -> Result<Vec<ItemWithMeta>, CoreError> {
+        let data_path = self.data_path.clone();
+        let db = self.db.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let conn = db.blocking_lock();
+            let item_service = ItemService::new(data_path);
+            item_service.list_items(&conn, &tags, &meta)
+        })
+        .await
+        .unwrap()
+    }
+
+    pub async fn delete_item(&self, id: i64) -> Result<(), CoreError> {
+        let data_path = self.data_path.clone();
+        let db = self.db.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let mut conn = db.blocking_lock();
+            let item_service = ItemService::new(data_path);
+            item_service.delete_item(&mut conn, id)
+        })
+        .await
+        .unwrap()
+    }
+
+    pub async fn save_item_from_mcp(
+        &self,
+        content: Vec<u8>,
+        tags: Vec<String>,
+        metadata: HashMap<String, String>,
+    ) -> Result<ItemWithMeta, CoreError> {
+        let data_path = self.data_path.clone();
+        let db = self.db.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let mut conn = db.blocking_lock();
+            let item_service = ItemService::new(data_path);
+            item_service.save_item_from_mcp(&content, &tags, &metadata, &mut conn)
+        })
+        .await
+        .unwrap()
+    }
+}

src/services/compression_service.rs

@@ -0,0 +1,33 @@
+use crate::compression_engine::{get_compression_engine, CompressionType};
+use crate::services::error::CoreError;
+use std::io::Read;
+use std::path::PathBuf;
+use std::str::FromStr;
+use anyhow::anyhow;
+
+pub struct CompressionService;
+
+impl CompressionService {
+    pub fn new() -> Self {
+        Self
+    }
+
+    pub fn get_item_content(&self, item_path: PathBuf, compression: &str) -> Result<Vec<u8>, CoreError> {
+        let compression_type = CompressionType::from_str(compression)
+            .map_err(|e| CoreError::Compression(e.to_string()))?;
+        let engine = get_compression_engine(compression_type)
+            .map_err(|e| CoreError::Other(anyhow!(e.to_string())))?;
+
+        let mut reader = engine.open(item_path.clone())
+            .map_err(|e| CoreError::Other(anyhow!("Failed to open item file {:?}: {}", item_path, e)))?;
+        let mut content = Vec::new();
+        reader.read_to_end(&mut content)?;
+        Ok(content)
+    }
+}
+
+impl Default for CompressionService {
+    fn default() -> Self {
+        Self::new()
+    }
+}

src/services/error.rs

@@ -0,0 +1,37 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum CoreError {
+    #[error("Database error: {0}")]
+    Database(#[from] rusqlite::Error),
+
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("Item not found with id {0}")]
+    ItemNotFound(i64),
+
+    #[error("Item not found")]
+    ItemNotFoundGeneric,
+
+    #[error("Invalid input: {0}")]
+    InvalidInput(String),
+
+    #[error("Configuration error: {0}")]
+    Config(String),
+
+    #[error("Compression error: {0}")]
+    Compression(String),
+
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
+impl From<rusqlite_migration::Error> for CoreError {
+    fn from(err: rusqlite_migration::Error) -> Self {
+        match err {
+            rusqlite_migration::Error::RusqliteError { err: e, .. } => CoreError::Database(e),
+            e => CoreError::Other(e.into()),
+        }
+    }
+}

src/services/meta_service.rs

@@ -0,0 +1,125 @@
+use crate::config::Settings;
+use crate::meta_plugin::{get_meta_plugin, MetaPlugin, MetaPluginType};
+use crate::modes::common::{settings_digest_type, settings_meta_plugin_types};
+use clap::Command;
+use log::debug;
+use rusqlite::Connection;
+use std::collections::HashMap;
+
+pub struct MetaService;
+
+impl MetaService {
+    pub fn new() -> Self {
+        Self
+    }
+
+    pub fn get_plugins(&self, cmd: &mut Command, settings: &Settings) -> Vec<Box<dyn MetaPlugin>> {
+        let mut meta_plugin_types: Vec<MetaPluginType> = settings_meta_plugin_types(cmd, settings);
+
+        let digest_type = settings_digest_type(cmd, settings);
+        let digest_meta_plugin_type = match digest_type {
+            MetaPluginType::DigestSha256 => Some(MetaPluginType::DigestSha256),
+            MetaPluginType::DigestMd5 => Some(MetaPluginType::DigestMd5),
+            _ => None,
+        };
+
+        if let Some(digest_plugin_type) = digest_meta_plugin_type {
+            if !meta_plugin_types.contains(&digest_plugin_type) {
+                meta_plugin_types.push(digest_plugin_type);
+            }
+        }
+
+        debug!("MetaService: Meta plugin types: {:?}", meta_plugin_types);
+
+        let mut meta_plugins: Vec<Box<dyn MetaPlugin>> = meta_plugin_types
+            .iter()
+            .map(|meta_plugin_type| get_meta_plugin(meta_plugin_type.clone()))
+            .collect();
+
+        if let Some(meta_plugin_configs) = &settings.meta_plugins {
+            for meta_plugin in meta_plugins.iter_mut() {
+                let plugin_name = meta_plugin.meta_name();
+                if let Some(config) = meta_plugin_configs.iter().find(|c| c.name == plugin_name) {
+                    let mut configured_outputs = meta_plugin.outputs().clone();
+                    for (key, value) in &config.outputs {
+                        configured_outputs.insert(key.clone(), serde_yaml::Value::String(value.clone()));
+                    }
+
+                    let mut configured_options = meta_plugin.default_options();
+                    for (key, value) in &config.options {
+                        configured_options.insert(key.clone(), value.clone());
+                    }
+
+                    if let Err(e) = meta_plugin.configure_outputs(&configured_outputs) {
+                        eprintln!("Warning: Failed to configure outputs for meta plugin '{}': {}", plugin_name, e);
+                    }
+
+                    if let Err(e) = meta_plugin.configure_options(&configured_options) {
+                        eprintln! (
+                            "Warning: Failed to configure options for meta plugin '{}': {}",
+                            plugin_name, e
+                        );
+                    }
+                }
+            }
+        }
+
+        let original_len = meta_plugins.len();
+        meta_plugins.retain(|meta_plugin| meta_plugin.is_supported());
+        if meta_plugins.len() < original_len {
+            // This is not perfect as it doesn't say which one, but avoids complex logic from save.rs
+            eprintln!("Warning: Some meta plugins are enabled but not supported on this system");
+        }
+
+        meta_plugins
+    }
+
+    pub fn initialize_plugins(
+        &self,
+        plugins: &mut [Box<dyn MetaPlugin>],
+        conn: &Connection,
+        item_id: i64,
+    ) {
+        for meta_plugin in plugins.iter_mut() {
+            if let Err(e) = meta_plugin.initialize(conn, item_id) {
+                eprintln!("Warning: Failed to initialize meta plugin: {}", e);
+            }
+        }
+    }
+
+    pub fn process_chunk(
+        &self,
+        plugins: &mut [Box<dyn MetaPlugin>],
+        chunk: &[u8],
+        conn: &Connection,
+    ) {
+        for meta_plugin in plugins.iter_mut() {
+            meta_plugin.update(chunk, conn);
+        }
+    }
+
+    pub fn finalize_plugins(&self, plugins: &mut [Box<dyn MetaPlugin>], conn: &Connection) {
+        for meta_plugin in plugins.iter_mut() {
+            if let Err(e) = meta_plugin.finalize(conn) {
+                eprintln!("Warning: Failed to finalize meta plugin: {}", e);
+            }
+        }
+    }
+
+    pub fn collect_initial_meta(&self) -> HashMap<String, String> {
+        let mut item_meta: HashMap<String, String> = crate::modes::common::get_meta_from_env();
+
+        if let Ok(hostname) = gethostname::gethostname().into_string() {
+            if !item_meta.contains_key("hostname") {
+                item_meta.insert("hostname".to_string(), hostname);
+            }
+        }
+        item_meta
+    }
+}
+
+impl Default for MetaService {
+    fn default() -> Self {
+        Self::new()
+    }
+}

src/services/mod.rs

@@ -0,0 +1,6 @@
+pub mod async_item_service;
+pub mod compression_service;
+pub mod error;
+pub mod item_service;
+pub mod meta_service;
+pub mod types;

src/services/types.rs

@@ -0,0 +1,22 @@
+use crate::db::{Item, Meta, Tag};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ItemWithMeta {
+    pub item: Item,
+    pub tags: Vec<Tag>,
+    pub meta: Vec<Meta>,
+}
+
+impl ItemWithMeta {
+    pub fn meta_as_map(&self) -> HashMap<String, String> {
+        self.meta.iter().cloned().map(|m| (m.name, m.value)).collect()
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ItemWithContent {
+    pub item_with_meta: ItemWithMeta,
+    pub content: Vec<u8>,
+}