keep/src/services/async_item_service.rs

use crate::services::error::CoreError;
use crate::services::item_service::ItemService;
use crate::services::types::{ItemWithContent, ItemWithMeta};
use crate::common::is_binary::is_binary;
use bytes::Bytes;
use rusqlite::Connection;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::io::{AsyncReadExt, AsyncSeekExt};
use tokio_util::io::ReaderStream;

/// 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 {
    pub data_dir: 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 stream_item_content(
        &self,
        item_id: i64,
        allow_binary: bool,
        offset: u64,
        length: u64,
    ) -> Result<(impl tokio_stream::Stream<Item = Result<bytes::Bytes, std::io::Error>>, String), CoreError> {
        let item_with_content = self.get_item_content(item_id).await?;
        let metadata = item_with_content.item_with_meta.meta_as_map();
        
        // Check if content is binary when allow_binary is false
        if !allow_binary {
            let is_content_binary = if let Some(binary_val) = metadata.get("binary") {
                binary_val == "true"
            } else {
                // If binary metadata not available, check the actual content
                is_binary(&item_with_content.content)
            };
            
            if is_content_binary {
                return Err(CoreError::InvalidInput("Binary content not allowed".to_string()));
            }
        }
        
        let mime_type = metadata
            .get("mime_type")
            .map(|s| s.to_string())
            .unwrap_or_else(|| "application/octet-stream".to_string());
        
        // Open the file for streaming
        let file_path = self.data_path.join(format!("{}.dat", item_id));
        let file = tokio::fs::File::open(&file_path).await?;
        let mut buffered_file = tokio::io::BufReader::new(file);
        
        // Seek to the requested offset if needed
        if offset > 0 {
            buffered_file.seek(std::io::SeekFrom::Start(offset)).await?;
        }
        
        // Create a reader stream with optional length limit
        // Create a reader stream with optional length limit
        let stream = if length > 0 {
            // Limit the stream to the specified length
            Box::pin(ReaderStream::new(buffered_file.take(length))) as std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<Bytes, std::io::Error>> + Send>>
        } else {
            // Stream the entire file from the offset
            Box::pin(ReaderStream::new(buffered_file)) as std::pin::Pin<Box<dyn tokio_stream::Stream<Item = Result<Bytes, std::io::Error>> + Send>>
        };
        
        Ok((stream, mime_type))
    }

    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()
    }
}