asp/keep
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: optimize item content streaming to reduce redundant database queries

Browse Source 
Co-authored-by: aider (openai/andrew/openrouter/qwen/qwen3-coder) <aider@aider.chat>
This commit is contained in:
Andrew Phillips
2025-08-25 20:17:45 -03:00
parent b9d6bd52d5
commit 4acec3d3dd
1 changed files with 20 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
src/services/async_item_service.rs
View File

@@ -63,15 +63,15 @@ impl AsyncItemService {
let db = self.db.clone(); let db = self.db.clone();
// Get item metadata first to check binary status and get MIME type // Get item metadata first to check binary status and get MIME type
let item_with_meta = tokio::task::spawn_blocking(move || { let item_with_content = tokio::task::spawn_blocking(move || {
let conn = db.blocking_lock(); let conn = db.blocking_lock();
let item_service = ItemService::new(data_dir); let item_service = ItemService::new(data_dir);
item_service.get_item(&conn, item_id) item_service.get_item_content(&conn, item_id)
}) })
.await .await
.unwrap()?; .unwrap()?;
let metadata = item_with_meta.meta_as_map(); let metadata = item_with_content.item_with_meta.meta_as_map();
let mime_type = metadata let mime_type = metadata
.get("mime_type") .get("mime_type")
.map(|s| s.to_string()) .map(|s| s.to_string())
@@ -84,19 +84,7 @@ impl AsyncItemService {
} else { } else {
// For the binary check, we need to read a sample of the content // For the binary check, we need to read a sample of the content
// We'll read the first 8192 bytes to determine if it's binary // We'll read the first 8192 bytes to determine if it's binary
let data_dir = self.data_dir.clone(); crate::common::is_binary::is_binary(&item_with_content.content)
let db = self.db.clone();
let sample_content = tokio::task::spawn_blocking(move || {
let conn = db.blocking_lock();
let item_service = ItemService::new(data_dir);
let item_with_content = item_service.get_item_content(&conn, item_id)?;
Ok::<Vec<u8>, CoreError>(item_with_content.content)
})
.await
.unwrap()?;
crate::common::is_binary::is_binary(&sample_content)
}; };
if is_content_binary { if is_content_binary {
@@ -105,42 +93,23 @@ impl AsyncItemService {
} }
// Create a stream that reads only the requested portion // Create a stream that reads only the requested portion
let data_dir = self.data_dir.clone(); let content = item_with_content.content;
let db = self.db.clone(); let content_len = content.len() as u64;
let stream = Box::pin(tokio_stream::wrappers::ReceiverStream::new({ // Apply offset and length constraints
let (tx, rx) = tokio::sync::mpsc::channel(1); let start = std::cmp::min(offset, content_len);
let end = if length > 0 {
std::cmp::min(start + length, content_len)
} else {
content_len
};
tokio::task::spawn_blocking(move || { let stream = if start < content_len {
let conn = db.blocking_lock(); let chunk = tokio_util::bytes::Bytes::from(content[start as usize..end as usize].to_vec());
let item_service = ItemService::new(data_dir); Box::pin(tokio_stream::iter(vec![Ok(chunk)]))
} else {
match item_service.get_item_content(&conn, item_id) { Box::pin(tokio_stream::iter(vec![]))
Ok(item_with_content) => { };
let content = item_with_content.content;
let content_len = content.len() as u64;
// Apply offset and length constraints
let start = std::cmp::min(offset, content_len);
let end = if length > 0 {
std::cmp::min(start + length, content_len)
} else {
content_len
};
if start < content_len {
let chunk = tokio_util::bytes::Bytes::from(content[start as usize..end as usize].to_vec());
let _ = tx.blocking_send(Ok(chunk));
}
}
Err(e) => {
let _ = tx.blocking_send(Err(std::io::Error::new(std::io::ErrorKind::Other, e.to_string())));
}
}
});
rx
}));
Ok((stream, mime_type)) Ok((stream, mime_type))
} }