commit ae9fb1278f141a962d420dbfff1f37452ff77d52
parent 8fd1574e6cc9c3fcc24d0dd2b91c131795d01f9a
Author: Silas Brack <silasbrack@gmail.com>
Date: Sun, 14 Jun 2026 13:54:17 +0200
refactor: dedicated async DB thread pool, remove spawn_blocking
Replace thread-local connections + spawn_blocking with dedicated
OS threads for reads (N=cores) and writes (1). Queries dispatch
via crossbeam channel, results via tokio oneshot. Fully async.
Structurally isolates DB from CPU-heavy work (argon2). No
semaphore, no max_blocking_threads cap, no thread-local RefCell.
Story page: 78k → 119k rps. Contention: 5.2k → 8.9k rps.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Diffstat:
13 files changed, 618 insertions(+), 605 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -229,6 +229,21 @@ dependencies = [
]
[[package]]
+name = "crossbeam-channel"
+version = "0.5.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2"
+dependencies = [
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
+
+[[package]]
name = "crypto-common"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -748,6 +763,7 @@ dependencies = [
"askama",
"axum",
"axum-extra",
+ "crossbeam-channel",
"rand",
"rusqlite",
"serde",
diff --git a/Cargo.toml b/Cargo.toml
@@ -13,3 +13,4 @@ tokio-stream = { version = "0.1", features = ["sync"] }
rusqlite = "0.31"
argon2 = "0.5"
rand = "0.8"
+crossbeam-channel = "0.5"
diff --git a/src/database.rs b/src/database.rs
@@ -1,6 +1,4 @@
use std::collections::HashMap;
-use std::sync::Mutex;
-
use rusqlite::{params, Connection};
use crate::error::AppError;
@@ -206,14 +204,8 @@ const PRAGMAS: &str = "\
PRAGMA temp_store=MEMORY;\
";
-pub struct Database {
- path: String,
- writer: Mutex<Connection>,
-}
-
-thread_local! {
- static READ_CONN: std::cell::RefCell<Option<Connection>> = const { std::cell::RefCell::new(None) };
-}
+type ReadJob = Box<dyn FnOnce(&Connection) + Send>;
+type WriteJob = Box<dyn FnOnce(&Connection) + Send>;
fn open_conn(path: &str) -> Result<Connection, AppError> {
let c = Connection::open(path)
@@ -223,6 +215,11 @@ fn open_conn(path: &str) -> Result<Connection, AppError> {
Ok(c)
}
+pub struct Database {
+ read_tx: crossbeam_channel::Sender<ReadJob>,
+ write_tx: std::sync::mpsc::Sender<WriteJob>,
+}
+
impl Database {
pub fn new(path: &str) -> Result<Self, AppError> {
if let Some(parent) = std::path::Path::new(path).parent() {
@@ -230,33 +227,62 @@ impl Database {
.map_err(|e| AppError::Database(format!("Failed to create db directory: {e}")))?;
}
- // Run migrations on the writer connection
- let writer = open_conn(path)?;
- crate::migrations::run(&writer).map_err(AppError::Database)?;
+ // Run migrations
+ let migration_conn = open_conn(path)?;
+ crate::migrations::run(&migration_conn).map_err(AppError::Database)?;
+ drop(migration_conn);
+
+ // Dedicated read pool: N threads, each owns a connection
+ let cores = std::thread::available_parallelism()
+ .map(|n| n.get())
+ .unwrap_or(4);
+ let (read_tx, read_rx) = crossbeam_channel::unbounded::<ReadJob>();
+ for _ in 0..cores {
+ let rx = read_rx.clone();
+ let p = path.to_string();
+ std::thread::spawn(move || {
+ let conn = open_conn(&p).expect("failed to open read connection");
+ while let Ok(job) = rx.recv() {
+ job(&conn);
+ }
+ });
+ }
- Ok(Self {
- path: path.to_string(),
- writer: Mutex::new(writer),
- })
+ // Dedicated writer: 1 thread, owns the write connection
+ let (write_tx, write_rx) = std::sync::mpsc::channel::<WriteJob>();
+ let p = path.to_string();
+ std::thread::spawn(move || {
+ let conn = open_conn(&p).expect("failed to open write connection");
+ while let Ok(job) = write_rx.recv() {
+ job(&conn);
+ }
+ });
+
+ Ok(Self { read_tx, write_tx })
}
- /// Run a read query on a thread-local connection. The connection
- /// never escapes the closure, so the borrow is sound.
- fn with_read<T>(&self, f: impl FnOnce(&Connection) -> Result<T, AppError>) -> Result<T, AppError> {
- READ_CONN.with(|cell| {
- let mut borrow = cell.borrow_mut();
- if borrow.is_none() {
- *borrow = Some(open_conn(&self.path)?);
- }
- f(borrow.as_ref().unwrap())
- })
+ /// Dispatch a read query to the read pool, await the result.
+ pub async fn with_read<T: Send + 'static>(
+ &self,
+ f: impl FnOnce(&Connection) -> Result<T, AppError> + Send + 'static,
+ ) -> Result<T, AppError> {
+ let (tx, rx) = tokio::sync::oneshot::channel();
+ self.read_tx.send(Box::new(move |conn| {
+ let _ = tx.send(f(conn));
+ })).map_err(|_| AppError::Database("read pool gone".into()))?;
+ rx.await.map_err(|_| AppError::Database("read cancelled".into()))?
}
- /// Get the exclusive writer connection.
- fn write_conn(&self) -> Result<std::sync::MutexGuard<'_, Connection>, AppError> {
- self.writer
- .lock()
- .map_err(|e| AppError::Database(format!("Writer lock poisoned: {e}")))
+ /// Dispatch a write to the single writer thread, await the result.
+ pub async fn with_write<T: Send + 'static>(
+ &self,
+ f: impl FnOnce(&Connection) -> Result<T, AppError> + Send + 'static,
+ ) -> Result<T, AppError> {
+ let (tx, rx) = tokio::sync::oneshot::channel();
+ self.write_tx.send(Box::new(move |conn| {
+ let _ = tx.send(f(conn));
+ })).map_err(|_| AppError::Database("writer gone".into()))?;
+ rx.await.map_err(|_| AppError::Database("write cancelled".into()))?
}
// ======================================================================
@@ -280,8 +306,8 @@ impl Database {
})
}
- pub fn get_stories_new(&self, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_stories_new(&self, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT * FROM story ORDER BY published DESC LIMIT ?1 OFFSET ?2"
).map_err(|e| AppError::Database(e.to_string()))?;
@@ -292,11 +318,11 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(stories)
- })
+ }).await
}
- pub fn get_stories_top(&self, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_stories_top(&self, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT * FROM story ORDER BY score DESC, published DESC LIMIT ?1 OFFSET ?2"
).map_err(|e| AppError::Database(e.to_string()))?;
@@ -307,12 +333,12 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(stories)
- })
+ }).await
}
- pub fn get_stories_hot(&self, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
+ pub async fn get_stories_hot(&self, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
let fetch_limit = (limit + offset) * 2;
- self.with_read(|conn| {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT * FROM story ORDER BY published DESC LIMIT ?1"
).map_err(|e| AppError::Database(e.to_string()))?;
@@ -330,11 +356,11 @@ impl Database {
let result: Vec<Story> = stories.into_iter().skip(offset).take(limit).collect();
Ok(result)
- })
+ }).await
}
- pub fn get_story(&self, id: i64) -> Result<Option<Story>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_story(&self, id: i64) -> Result<Option<Story>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached("SELECT * FROM story WHERE id = ?1")
.map_err(|e| AppError::Database(e.to_string()))?;
@@ -346,10 +372,10 @@ impl Database {
Some(Err(e)) => Err(AppError::Database(e.to_string())),
None => Ok(None),
}
- })
+ }).await
}
- pub fn create_story(
+ pub async fn create_story(
&self,
user_id: i64,
title: &str,
@@ -358,21 +384,26 @@ impl Database {
domain: Option<&str>,
language: &str,
) -> Result<Story, AppError> {
- let conn = self.write_conn()?;
- let published = now_epoch();
-
- conn.execute(
- "INSERT INTO story (title, url, text, domain, published, language, created_by)
- VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
- params![title, url, text, domain, published, language, user_id],
- ).map_err(|e| AppError::Database(e.to_string()))?;
-
- let id = conn.last_insert_rowid();
- drop(conn);
- self.get_story(id)?.ok_or_else(|| AppError::Database("Failed to retrieve created story".to_string()))
+ let title = title.to_string();
+ let url = url.map(|s| s.to_string());
+ let text = text.to_string();
+ let domain = domain.map(|s| s.to_string());
+ let language = language.to_string();
+ let id = self.with_write(move |conn| {
+ let published = now_epoch();
+
+ conn.execute(
+ "INSERT INTO story (title, url, text, domain, published, language, created_by)
+ VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
+ params![title, url, text, domain, published, language, user_id],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+
+ Ok(conn.last_insert_rowid())
+ }).await?;
+ self.get_story(id).await?.ok_or_else(|| AppError::Database("Failed to retrieve created story".to_string()))
}
- pub fn update_story(
+ pub async fn update_story(
&self,
id: i64,
title: &str,
@@ -380,16 +411,21 @@ impl Database {
text: &str,
domain: Option<&str>,
) -> Result<(), AppError> {
- let conn = self.write_conn()?;
- conn.execute(
- "UPDATE story SET title = ?1, url = ?2, text = ?3, domain = ?4 WHERE id = ?5",
- params![title, url, text, domain, id],
- ).map_err(|e| AppError::Database(e.to_string()))?;
- Ok(())
+ let title = title.to_string();
+ let url = url.map(|s| s.to_string());
+ let text = text.to_string();
+ let domain = domain.map(|s| s.to_string());
+ self.with_write(move |conn| {
+ conn.execute(
+ "UPDATE story SET title = ?1, url = ?2, text = ?3, domain = ?4 WHERE id = ?5",
+ params![title, url, text, domain, id],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(())
+ }).await
}
- pub fn get_stories_by_tag(&self, category_id: i64, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_stories_by_tag(&self, category_id: i64, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT s.* FROM story s
JOIN story_category sc ON sc.news_item_id = s.id
@@ -404,11 +440,11 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(stories)
- })
+ }).await
}
- pub fn get_stories_by_user(&self, user_id: i64, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_stories_by_user(&self, user_id: i64, limit: usize, offset: usize) -> Result<Vec<Story>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT * FROM story WHERE created_by = ?1 ORDER BY published DESC LIMIT ?2 OFFSET ?3"
).map_err(|e| AppError::Database(e.to_string()))?;
@@ -419,19 +455,20 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(stories)
- })
+ }).await
}
/// Search stories with selectivity check. Terms appearing in more than
/// `max_doc_ratio` of all documents are rejected — they produce low-quality
/// results and are expensive to rank.
- pub fn search_stories(
+ pub async fn search_stories(
&self,
query: &str,
limit: usize,
max_doc_ratio: f64,
) -> Result<Vec<SearchResult>, AppError> {
- self.with_read(|conn| {
+ let query = query.to_string();
+ self.with_read(move |conn| {
// Check selectivity: reject terms that match too many documents
let total_docs: i64 = conn
.query_row("SELECT COUNT(*) FROM story", [], |row| row.get(0))
@@ -476,7 +513,7 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(results)
- })
+ }).await
}
// ======================================================================
@@ -498,8 +535,8 @@ impl Database {
})
}
- pub fn get_comments_for_story(&self, story_id: i64) -> Result<Vec<Comment>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_comments_for_story(&self, story_id: i64) -> Result<Vec<Comment>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT * FROM comment WHERE story_id = ?1 ORDER BY path"
).map_err(|e| AppError::Database(e.to_string()))?;
@@ -510,72 +547,72 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(comments)
- })
+ }).await
}
- pub fn create_comment(
+ pub async fn create_comment(
&self,
user_id: i64,
story_id: i64,
parent_id: Option<i64>,
text: &str,
) -> Result<Comment, AppError> {
- let conn = self.write_conn()?;
-
- // Calculate path and depth
- let (path, depth) = match parent_id {
- Some(pid) => {
- let parent: Option<(String, i64)> = conn.query_row(
- "SELECT path, depth FROM comment WHERE id = ?1",
- params![pid],
- |row| Ok((row.get(0)?, row.get(1)?)),
- ).ok();
-
- match parent {
- Some((parent_path, parent_depth)) => {
- let sibling_count: i64 = conn.query_row(
- "SELECT COUNT(*) FROM comment WHERE parent_id = ?1",
- params![pid],
- |row| row.get(0),
- ).map_err(|e| AppError::Database(e.to_string()))?;
-
- let path = format!("{}.{}", parent_path, sibling_count + 1);
- (path, parent_depth + 1)
- }
- None => {
- // Parent not found, treat as root
- let root_count: i64 = conn.query_row(
- "SELECT COUNT(*) FROM comment WHERE story_id = ?1 AND depth = 0",
- params![story_id],
- |row| row.get(0),
- ).map_err(|e| AppError::Database(e.to_string()))?;
- ((root_count + 1).to_string(), 0)
+ let text = text.to_string();
+ let id = self.with_write(move |conn| {
+ // Calculate path and depth
+ let (path, depth) = match parent_id {
+ Some(pid) => {
+ let parent: Option<(String, i64)> = conn.query_row(
+ "SELECT path, depth FROM comment WHERE id = ?1",
+ params![pid],
+ |row| Ok((row.get(0)?, row.get(1)?)),
+ ).ok();
+
+ match parent {
+ Some((parent_path, parent_depth)) => {
+ let sibling_count: i64 = conn.query_row(
+ "SELECT COUNT(*) FROM comment WHERE parent_id = ?1",
+ params![pid],
+ |row| row.get(0),
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+
+ let path = format!("{}.{}", parent_path, sibling_count + 1);
+ (path, parent_depth + 1)
+ }
+ None => {
+ // Parent not found, treat as root
+ let root_count: i64 = conn.query_row(
+ "SELECT COUNT(*) FROM comment WHERE story_id = ?1 AND depth = 0",
+ params![story_id],
+ |row| row.get(0),
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ ((root_count + 1).to_string(), 0)
+ }
}
}
- }
- None => {
- let root_count: i64 = conn.query_row(
- "SELECT COUNT(*) FROM comment WHERE story_id = ?1 AND depth = 0",
- params![story_id],
- |row| row.get(0),
- ).map_err(|e| AppError::Database(e.to_string()))?;
- ((root_count + 1).to_string(), 0)
- }
- };
+ None => {
+ let root_count: i64 = conn.query_row(
+ "SELECT COUNT(*) FROM comment WHERE story_id = ?1 AND depth = 0",
+ params![story_id],
+ |row| row.get(0),
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ ((root_count + 1).to_string(), 0)
+ }
+ };
- conn.execute(
- "INSERT INTO comment (story_id, parent_id, path, depth, text, created_by, created_at)
- VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
- params![story_id, parent_id, path, depth, text, user_id, now_epoch()],
- ).map_err(|e| AppError::Database(e.to_string()))?;
+ conn.execute(
+ "INSERT INTO comment (story_id, parent_id, path, depth, text, created_by, created_at)
+ VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
+ params![story_id, parent_id, path, depth, text, user_id, now_epoch()],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
- let id = conn.last_insert_rowid();
- drop(conn);
- self.get_comment(id)?.ok_or_else(|| AppError::Database("Failed to retrieve created comment".to_string()))
+ Ok(conn.last_insert_rowid())
+ }).await?;
+ self.get_comment(id).await?.ok_or_else(|| AppError::Database("Failed to retrieve created comment".to_string()))
}
- pub fn get_comment(&self, id: i64) -> Result<Option<Comment>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_comment(&self, id: i64) -> Result<Option<Comment>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached("SELECT * FROM comment WHERE id = ?1")
.map_err(|e| AppError::Database(e.to_string()))?;
@@ -587,11 +624,11 @@ impl Database {
Some(Err(e)) => Err(AppError::Database(e.to_string())),
None => Ok(None),
}
- })
+ }).await
}
- pub fn get_comments_by_user(&self, user_id: i64, limit: usize, offset: usize) -> Result<Vec<Comment>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_comments_by_user(&self, user_id: i64, limit: usize, offset: usize) -> Result<Vec<Comment>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT * FROM comment WHERE created_by = ?1 ORDER BY created_at DESC LIMIT ?2 OFFSET ?3"
).map_err(|e| AppError::Database(e.to_string()))?;
@@ -602,7 +639,7 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(comments)
- })
+ }).await
}
// ======================================================================
@@ -610,37 +647,38 @@ impl Database {
// ======================================================================
/// Toggle vote: INSERT if not exists, DELETE if exists. Returns true if now voted.
- pub fn toggle_vote(&self, user_id: i64, target_type: i64, target_id: i64) -> Result<bool, AppError> {
- let conn = self.write_conn()?;
-
- // Try to insert
- let result = conn.execute(
- "INSERT INTO vote (user_id, target_type, target_id, created_at) VALUES (?1, ?2, ?3, ?4)",
- params![user_id, target_type, target_id, now_epoch()],
- );
-
- match result {
- Ok(_) => Ok(true),
- Err(rusqlite::Error::SqliteFailure(err, _))
- if err.code == rusqlite::ffi::ErrorCode::ConstraintViolation =>
- {
- // Already voted, remove it
- conn.execute(
- "DELETE FROM vote WHERE user_id = ?1 AND target_type = ?2 AND target_id = ?3",
- params![user_id, target_type, target_id],
- ).map_err(|e| AppError::Database(e.to_string()))?;
- Ok(false)
+ pub async fn toggle_vote(&self, user_id: i64, target_type: i64, target_id: i64) -> Result<bool, AppError> {
+ self.with_write(move |conn| {
+ // Try to insert
+ let result = conn.execute(
+ "INSERT INTO vote (user_id, target_type, target_id, created_at) VALUES (?1, ?2, ?3, ?4)",
+ params![user_id, target_type, target_id, now_epoch()],
+ );
+
+ match result {
+ Ok(_) => Ok(true),
+ Err(rusqlite::Error::SqliteFailure(err, _))
+ if err.code == rusqlite::ffi::ErrorCode::ConstraintViolation =>
+ {
+ // Already voted, remove it
+ conn.execute(
+ "DELETE FROM vote WHERE user_id = ?1 AND target_type = ?2 AND target_id = ?3",
+ params![user_id, target_type, target_id],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(false)
+ }
+ Err(e) => Err(AppError::Database(e.to_string())),
}
- Err(e) => Err(AppError::Database(e.to_string())),
- }
+ }).await
}
- pub fn get_user_votes(&self, user_id: i64, target_type: i64, target_ids: &[i64]) -> Result<Vec<i64>, AppError> {
+ pub async fn get_user_votes(&self, user_id: i64, target_type: i64, target_ids: &[i64]) -> Result<Vec<i64>, AppError> {
if target_ids.is_empty() {
return Ok(vec![]);
}
- self.with_read(|conn| {
+ let target_ids = target_ids.to_vec();
+ self.with_read(move |conn| {
// Build placeholders for IN clause
let placeholders: Vec<String> = target_ids.iter().enumerate().map(|(i, _)| format!("?{}", i + 3)).collect();
let sql = format!(
@@ -654,8 +692,8 @@ impl Database {
let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
param_values.push(Box::new(user_id));
param_values.push(Box::new(target_type));
- for id in target_ids {
- param_values.push(Box::new(*id));
+ for &id in &target_ids {
+ param_values.push(Box::new(id));
}
let params_ref: Vec<&dyn rusqlite::types::ToSql> = param_values.iter().map(|p| p.as_ref()).collect();
@@ -665,15 +703,15 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(voted_ids)
- })
+ }).await
}
// ======================================================================
// Category queries
// ======================================================================
- pub fn get_categories(&self, limit: usize) -> Result<Vec<Category>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_categories(&self, limit: usize) -> Result<Vec<Category>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached("SELECT id, name FROM category LIMIT ?1")
.map_err(|e| AppError::Database(e.to_string()))?;
@@ -688,11 +726,11 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(categories)
- })
+ }).await
}
- pub fn get_categories_for_story(&self, story_id: i64) -> Result<Vec<Category>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_categories_for_story(&self, story_id: i64) -> Result<Vec<Category>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached(
"SELECT c.id, c.name FROM category c
JOIN story_category sc ON sc.category_id = c.id
@@ -710,25 +748,27 @@ impl Database {
.map_err(|e| AppError::Database(e.to_string()))?;
Ok(categories)
- })
+ }).await
}
- pub fn add_tag_to_story(&self, story_id: i64, category_id: i64) -> Result<(), AppError> {
- let conn = self.write_conn()?;
- conn.execute(
- "INSERT INTO story_category (news_item_id, category_id) VALUES (?1, ?2)",
- params![story_id, category_id],
- ).map_err(|e| AppError::Database(e.to_string()))?;
- Ok(())
+ pub async fn add_tag_to_story(&self, story_id: i64, category_id: i64) -> Result<(), AppError> {
+ self.with_write(move |conn| {
+ conn.execute(
+ "INSERT INTO story_category (news_item_id, category_id) VALUES (?1, ?2)",
+ params![story_id, category_id],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(())
+ }).await
}
- pub fn remove_all_tags_from_story(&self, story_id: i64) -> Result<(), AppError> {
- let conn = self.write_conn()?;
- conn.execute(
- "DELETE FROM story_category WHERE news_item_id = ?1",
- params![story_id],
- ).map_err(|e| AppError::Database(e.to_string()))?;
- Ok(())
+ pub async fn remove_all_tags_from_story(&self, story_id: i64) -> Result<(), AppError> {
+ self.with_write(move |conn| {
+ conn.execute(
+ "DELETE FROM story_category WHERE news_item_id = ?1",
+ params![story_id],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(())
+ }).await
}
// ======================================================================
@@ -750,8 +790,8 @@ impl Database {
})
}
- pub fn get_user_by_id(&self, id: i64) -> Result<Option<User>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_user_by_id(&self, id: i64) -> Result<Option<User>, AppError> {
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached("SELECT * FROM user WHERE id = ?1")
.map_err(|e| AppError::Database(e.to_string()))?;
@@ -763,11 +803,12 @@ impl Database {
Some(Err(e)) => Err(AppError::Database(e.to_string())),
None => Ok(None),
}
- })
+ }).await
}
- pub fn get_user_by_email(&self, email: &str) -> Result<Option<User>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_user_by_email(&self, email: &str) -> Result<Option<User>, AppError> {
+ let email = email.to_string();
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached("SELECT * FROM user WHERE email = ?1")
.map_err(|e| AppError::Database(e.to_string()))?;
@@ -779,11 +820,12 @@ impl Database {
Some(Err(e)) => Err(AppError::Database(e.to_string())),
None => Ok(None),
}
- })
+ }).await
}
- pub fn get_user_by_username(&self, username: &str) -> Result<Option<User>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_user_by_username(&self, username: &str) -> Result<Option<User>, AppError> {
+ let username = username.to_string();
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached("SELECT * FROM user WHERE username = ?1")
.map_err(|e| AppError::Database(e.to_string()))?;
@@ -795,59 +837,69 @@ impl Database {
Some(Err(e)) => Err(AppError::Database(e.to_string())),
None => Ok(None),
}
- })
+ }).await
}
- pub fn create_user(&self, email: &str, password_hash: &str, username: &str, verification_token: Option<&str>) -> Result<User, AppError> {
- let conn = self.write_conn()?;
- conn.execute(
- "INSERT INTO user (email, password_hash, username, verification_token, created_at) VALUES (?1, ?2, ?3, ?4, ?5)",
- params![email, password_hash, username, verification_token, now_epoch()],
- ).map_err(|e| AppError::Database(e.to_string()))?;
+ pub async fn create_user(&self, email: &str, password_hash: &str, username: &str, verification_token: Option<&str>) -> Result<User, AppError> {
+ let email = email.to_string();
+ let password_hash = password_hash.to_string();
+ let username = username.to_string();
+ let verification_token = verification_token.map(|s| s.to_string());
+ let id = self.with_write(move |conn| {
+ conn.execute(
+ "INSERT INTO user (email, password_hash, username, verification_token, created_at) VALUES (?1, ?2, ?3, ?4, ?5)",
+ params![email, password_hash, username, verification_token, now_epoch()],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
- let id = conn.last_insert_rowid();
- drop(conn);
- self.get_user_by_id(id)?.ok_or_else(|| AppError::Database("Failed to retrieve created user".to_string()))
+ Ok(conn.last_insert_rowid())
+ }).await?;
+ self.get_user_by_id(id).await?.ok_or_else(|| AppError::Database("Failed to retrieve created user".to_string()))
}
- pub fn verify_email_token(&self, token: &str) -> Result<bool, AppError> {
- let conn = self.write_conn()?;
- let rows_affected = conn.execute(
- "UPDATE user SET email_verified = 1, verification_token = NULL WHERE verification_token = ?1",
- params![token],
- ).map_err(|e| AppError::Database(e.to_string()))?;
- Ok(rows_affected > 0)
+ pub async fn verify_email_token(&self, token: &str) -> Result<bool, AppError> {
+ let token = token.to_string();
+ self.with_write(move |conn| {
+ let rows_affected = conn.execute(
+ "UPDATE user SET email_verified = 1, verification_token = NULL WHERE verification_token = ?1",
+ params![token],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(rows_affected > 0)
+ }).await
}
- pub fn update_verification_token(&self, user_id: i64, token: &str) -> Result<(), AppError> {
- let conn = self.write_conn()?;
- conn.execute(
- "UPDATE user SET verification_token = ?1 WHERE id = ?2",
- params![token, user_id],
- ).map_err(|e| AppError::Database(e.to_string()))?;
- Ok(())
+ pub async fn update_verification_token(&self, user_id: i64, token: &str) -> Result<(), AppError> {
+ let token = token.to_string();
+ self.with_write(move |conn| {
+ conn.execute(
+ "UPDATE user SET verification_token = ?1 WHERE id = ?2",
+ params![token, user_id],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(())
+ }).await
}
// ======================================================================
// Session queries
// ======================================================================
- pub fn create_session(&self, user_id: i64) -> Result<String, AppError> {
+ pub async fn create_session(&self, user_id: i64) -> Result<String, AppError> {
use rand::Rng;
let bytes: [u8; 32] = rand::thread_rng().gen();
let token: String = bytes.iter().map(|b| format!("{:02x}", b)).collect();
- let conn = self.write_conn()?;
- conn.execute(
- "INSERT INTO session (token, user_id, created_at) VALUES (?1, ?2, ?3)",
- params![token, user_id, now_epoch()],
- ).map_err(|e| AppError::Database(e.to_string()))?;
+ self.with_write(move |conn| {
+ conn.execute(
+ "INSERT INTO session (token, user_id, created_at) VALUES (?1, ?2, ?3)",
+ params![token, user_id, now_epoch()],
+ ).map_err(|e| AppError::Database(e.to_string()))?;
- Ok(token)
+ Ok(token)
+ }).await
}
- pub fn get_session_user_id(&self, token: &str) -> Result<Option<i64>, AppError> {
- self.with_read(|conn| {
+ pub async fn get_session_user_id(&self, token: &str) -> Result<Option<i64>, AppError> {
+ let token = token.to_string();
+ self.with_read(move |conn| {
let mut stmt = conn.prepare_cached("SELECT user_id FROM session WHERE token = ?1")
.map_err(|e| AppError::Database(e.to_string()))?;
@@ -856,33 +908,37 @@ impl Database {
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(AppError::Database(e.to_string())),
}
- })
+ }).await
}
- pub fn delete_session(&self, token: &str) -> Result<(), AppError> {
- let conn = self.write_conn()?;
- conn.execute("DELETE FROM session WHERE token = ?1", params![token])
- .map_err(|e| AppError::Database(e.to_string()))?;
- Ok(())
+ pub async fn delete_session(&self, token: &str) -> Result<(), AppError> {
+ let token = token.to_string();
+ self.with_write(move |conn| {
+ conn.execute("DELETE FROM session WHERE token = ?1", params![token])
+ .map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(())
+ }).await
}
- pub fn delete_user_sessions(&self, user_id: i64) -> Result<(), AppError> {
- let conn = self.write_conn()?;
- conn.execute("DELETE FROM session WHERE user_id = ?1", params![user_id])
- .map_err(|e| AppError::Database(e.to_string()))?;
- Ok(())
+ pub async fn delete_user_sessions(&self, user_id: i64) -> Result<(), AppError> {
+ self.with_write(move |conn| {
+ conn.execute("DELETE FROM session WHERE user_id = ?1", params![user_id])
+ .map_err(|e| AppError::Database(e.to_string()))?;
+ Ok(())
+ }).await
}
// ======================================================================
// Enrichment queries
// ======================================================================
- pub fn enrich_stories(&self, story_ids: &[i64]) -> Result<Vec<StoryMeta>, AppError> {
+ pub async fn enrich_stories(&self, story_ids: &[i64]) -> Result<Vec<StoryMeta>, AppError> {
if story_ids.is_empty() {
return Ok(vec![]);
}
- self.with_read(|conn| {
+ let story_ids = story_ids.to_vec();
+ self.with_read(move |conn| {
// Fetch author usernames
let placeholders: Vec<String> = story_ids.iter().enumerate().map(|(i, _)| format!("?{}", i + 1)).collect();
let sql = format!(
@@ -894,8 +950,8 @@ impl Database {
let mut stmt = conn.prepare_cached(&sql).map_err(|e| AppError::Database(e.to_string()))?;
let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
- for id in story_ids {
- param_values.push(Box::new(*id));
+ for &id in &story_ids {
+ param_values.push(Box::new(id));
}
let params_ref: Vec<&dyn rusqlite::types::ToSql> = param_values.iter().map(|p| p.as_ref()).collect();
@@ -940,15 +996,16 @@ impl Database {
.collect();
Ok(result)
- })
+ }).await
}
- pub fn enrich_comment_authors(&self, user_ids: &[i64]) -> Result<HashMap<i64, String>, AppError> {
+ pub async fn enrich_comment_authors(&self, user_ids: &[i64]) -> Result<HashMap<i64, String>, AppError> {
if user_ids.is_empty() {
return Ok(HashMap::new());
}
- self.with_read(|conn| {
+ let user_ids = user_ids.to_vec();
+ self.with_read(move |conn| {
let placeholders: Vec<String> = user_ids.iter().enumerate().map(|(i, _)| format!("?{}", i + 1)).collect();
let sql = format!(
"SELECT id, username FROM user WHERE id IN ({})",
@@ -957,8 +1014,8 @@ impl Database {
let mut stmt = conn.prepare_cached(&sql).map_err(|e| AppError::Database(e.to_string()))?;
let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
- for id in user_ids {
- param_values.push(Box::new(*id));
+ for &id in &user_ids {
+ param_values.push(Box::new(id));
}
let params_ref: Vec<&dyn rusqlite::types::ToSql> = param_values.iter().map(|p| p.as_ref()).collect();
@@ -971,21 +1028,21 @@ impl Database {
.collect();
Ok(map)
- })
+ }).await
}
- pub fn get_story_score(&self, id: i64) -> Result<i64, AppError> {
- self.with_read(|conn| {
+ pub async fn get_story_score(&self, id: i64) -> Result<i64, AppError> {
+ self.with_read(move |conn| {
conn.query_row("SELECT score FROM story WHERE id = ?1", params![id], |row| row.get(0))
.map_err(|e| AppError::Database(e.to_string()))
- })
+ }).await
}
- pub fn get_comment_score(&self, id: i64) -> Result<i64, AppError> {
- self.with_read(|conn| {
+ pub async fn get_comment_score(&self, id: i64) -> Result<i64, AppError> {
+ self.with_read(move |conn| {
conn.query_row("SELECT score FROM comment WHERE id = ?1", params![id], |row| row.get(0))
.map_err(|e| AppError::Database(e.to_string()))
- })
+ }).await
}
}
diff --git a/src/handlers/auth.rs b/src/handlers/auth.rs
@@ -130,10 +130,9 @@ pub async fn login(
let password = form.password.clone();
// Look up user (DB read — uses read pool)
- let user = state.db_query(move |db| {
- db.get_user_by_email(&email)?
- .ok_or_else(|| AppError::Unauthorized("Invalid email or password".to_string()))
- }).await.map_err(|e| render_login_error(&format!("{e}")))?;
+ let user = state.db().get_user_by_email(&email).await
+ .and_then(|opt| opt.ok_or_else(|| AppError::Unauthorized("Invalid email or password".to_string())))
+ .map_err(|e| render_login_error(&format!("{e}")))?;
if !user.email_verified {
return Err(render_login_error(
@@ -141,26 +140,21 @@ pub async fn login(
));
}
- // Verify password (CPU-bound — acquire semaphore to avoid starving DB reads)
+ // Verify password (CPU-bound — runs on tokio's blocking pool,
+ // which is separate from the dedicated DB thread pool)
let hash = user.password_hash.clone();
- let _permit = state.hash_semaphore.acquire().await
- .map_err(|e| render_login_error(&format!("{e}")))?;
let valid = tokio::task::spawn_blocking(move || {
auth::verify_password(&password, &hash)
}).await
.map_err(|e| render_login_error(&format!("{e}")))?
.map_err(|e| render_login_error(&format!("{e}")))?;
- drop(_permit);
if !valid {
return Err(render_login_error("Invalid email or password"));
}
// Create session (DB write)
- let user_id = user.id;
- let result = state.db_query(move |db| {
- db.create_session(user_id)
- }).await;
+ let result = state.db().create_session(user.id).await;
match result {
Ok(token) => {
@@ -221,32 +215,32 @@ pub async fn register(
let token_clone = token.clone();
// Check uniqueness (DB reads)
- let uname = username.clone();
- let em = email.clone();
- state.db_query(move |db| {
- if db.get_user_by_username(&uname)?.is_some() {
- return Err(AppError::Database("Username is already taken".to_string()));
- }
- if db.get_user_by_email(&em)?.is_some() {
- return Err(AppError::Database("Email is already registered".to_string()));
- }
- Ok(())
- }).await.map_err(|e| render_register_error(&form, &format!("{e}")))?;
+ let db = state.db();
+ if db.get_user_by_username(&username).await
+ .map_err(|e| render_register_error(&form, &format!("{e}")))?
+ .is_some()
+ {
+ return Err(render_register_error(&form, "Username is already taken"));
+ }
+ if db.get_user_by_email(&email).await
+ .map_err(|e| render_register_error(&form, &format!("{e}")))?
+ .is_some()
+ {
+ return Err(render_register_error(&form, "Email is already registered"));
+ }
// Hash password (CPU-bound — uses semaphore)
- let _permit = state.hash_semaphore.acquire().await
- .map_err(|e| render_register_error(&form, &format!("{e}")))?;
+ // Hash password (CPU-bound — runs on tokio's blocking pool,
+ // separate from DB thread pool)
let password_hash = tokio::task::spawn_blocking(move || {
auth::hash_password(&password)
}).await
.map_err(|e| render_register_error(&form, &format!("{e}")))?
.map_err(|e| render_register_error(&form, &format!("{e}")))?;
- drop(_permit);
// Create user (DB write)
- let result = state.db_query(move |db| {
- db.create_user(&email, &password_hash, &username, Some(&token_clone))
- }).await.map(|_| ());
+ let result = state.db().create_user(&email, &password_hash, &username, Some(&token_clone))
+ .await.map(|_| ());
match result {
Ok(()) => {
@@ -281,9 +275,7 @@ pub async fn verify_email(
) -> Result<Redirect, Html<String>> {
let token = query.token;
- let result = state.db_query(move |db| {
- db.verify_email_token(&token)
- }).await;
+ let result = state.db().verify_email_token(&token).await;
match result {
Ok(true) => Ok(Redirect::to("/login")),
@@ -338,17 +330,16 @@ pub async fn resend_verification(
};
let new_token_clone = new_token.clone();
- let user_found = state.db_query(move |db| {
- let user = db.get_user_by_email(&email)?;
- match user {
- Some(u) if !u.email_verified => {
- db.update_verification_token(u.id, &new_token_clone)?;
- Ok(true)
- }
- Some(_) => Ok(false), // already verified
- None => Ok(false), // user not found
+ let db = state.db();
+ let user = db.get_user_by_email(&email).await?;
+ let user_found = match user {
+ Some(u) if !u.email_verified => {
+ db.update_verification_token(u.id, &new_token_clone).await?;
+ true
}
- }).await?;
+ Some(_) => false, // already verified
+ None => false, // user not found
+ };
if user_found {
if let Err(e) = send_verification_email(&state, &form.email, &new_token) {
@@ -372,7 +363,7 @@ pub async fn logout(
jar: CookieJar,
) -> (CookieJar, Redirect) {
if let Some(token) = get_session_token(&jar) {
- let _ = state.db_query(move |db| db.delete_session(&token)).await;
+ let _ = state.db().delete_session(&token).await;
}
let jar = clear_auth_cookie(jar);
(jar, Redirect::to("/"))
diff --git a/src/handlers/comment.rs b/src/handlers/comment.rs
@@ -34,9 +34,8 @@ pub async fn create_comment(
let text = form.text.trim().to_string();
let comment_tx = state.comment_tx.clone();
- state.db_query(move |db| {
- db.create_comment(user_id, story_id, None, &text)
- }).await.map_err(|e| Html(format!("Failed to create comment: {}", e)))?;
+ state.db().create_comment(user_id, story_id, None, &text)
+ .await.map_err(|e| Html(format!("Failed to create comment: {}", e)))?;
// Notify SSE subscribers
let _ = comment_tx.send(story_id);
@@ -58,21 +57,19 @@ pub async fn show_reply_form(
));
}
- let (parent_comment, parent_author) = state.db_query(move |db| {
- let comment = db.get_comment(comment_id)?
- .ok_or_else(|| AppError::NotFound("Comment not found".to_string()))?;
-
- let author = if let Ok(uid) = comment.created_by.parse::<i64>() {
- db.get_user_by_id(uid)
- .ok()
- .flatten()
- .map(|u| u.username)
- } else {
- None
- };
-
- Ok((comment, author))
- }).await?;
+ let db = state.db();
+ let parent_comment = db.get_comment(comment_id).await?
+ .ok_or_else(|| AppError::NotFound("Comment not found".to_string()))?;
+
+ let parent_author = if let Ok(uid) = parent_comment.created_by.parse::<i64>() {
+ db.get_user_by_id(uid)
+ .await
+ .ok()
+ .flatten()
+ .map(|u| u.username)
+ } else {
+ None
+ };
let reply_form = ReplyFormTemplate {
comment_id,
@@ -113,9 +110,8 @@ pub async fn submit_reply(
let story_id = form.story_id;
let comment_tx = state.comment_tx.clone();
- state.db_query(move |db| {
- db.create_comment(user_id, story_id, Some(parent_id), &text)
- }).await.map_err(|e| Html(format!("Failed to create reply: {}", e)))?;
+ state.db().create_comment(user_id, story_id, Some(parent_id), &text)
+ .await.map_err(|e| Html(format!("Failed to create reply: {}", e)))?;
// Notify SSE subscribers
let _ = comment_tx.send(story_id);
diff --git a/src/handlers/feed.rs b/src/handlers/feed.rs
@@ -45,32 +45,30 @@ async fn fetch_feed(
let tag_id = query.tag;
let page = query.page;
- let (stories, categories, voted_ids, meta) = state.db_query(move |db| {
- let categories = db.get_categories(20)?;
-
- let all_stories = match tag_id {
- Some(tid) => db.get_stories_by_tag(tid, PAGE_LIMIT + 1, offset)?,
- None => match sort {
- FeedSort::Hot => db.get_stories_hot(PAGE_LIMIT + 1, offset)?,
- FeedSort::New => db.get_stories_new(PAGE_LIMIT + 1, offset)?,
- FeedSort::Top => db.get_stories_top(PAGE_LIMIT + 1, offset)?,
- },
- };
-
- let stories: Vec<_> = all_stories.into_iter().take(PAGE_LIMIT).collect();
- let story_ids: Vec<i64> = stories.iter().map(|s| s.id).collect();
-
- let voted_ids = if let Some(uid) = user_id {
- db.get_user_votes(uid, TARGET_TYPE_STORY, &story_ids)
- .unwrap_or_default()
- } else {
- vec![]
- };
-
- let meta = db.enrich_stories(&story_ids).unwrap_or_default();
-
- Ok((stories, categories, voted_ids, meta))
- }).await?;
+ let db = state.db();
+ let categories = db.get_categories(20).await?;
+
+ let all_stories = match tag_id {
+ Some(tid) => db.get_stories_by_tag(tid, PAGE_LIMIT + 1, offset).await?,
+ None => match sort {
+ FeedSort::Hot => db.get_stories_hot(PAGE_LIMIT + 1, offset).await?,
+ FeedSort::New => db.get_stories_new(PAGE_LIMIT + 1, offset).await?,
+ FeedSort::Top => db.get_stories_top(PAGE_LIMIT + 1, offset).await?,
+ },
+ };
+
+ let stories: Vec<_> = all_stories.into_iter().take(PAGE_LIMIT).collect();
+ let story_ids: Vec<i64> = stories.iter().map(|s| s.id).collect();
+
+ let voted_ids = if let Some(uid) = user_id {
+ db.get_user_votes(uid, TARGET_TYPE_STORY, &story_ids)
+ .await
+ .unwrap_or_default()
+ } else {
+ vec![]
+ };
+
+ let meta = db.enrich_stories(&story_ids).await.unwrap_or_default();
let reached_end = stories.len() <= PAGE_LIMIT;
diff --git a/src/handlers/search.rs b/src/handlers/search.rs
@@ -37,10 +37,7 @@ pub async fn search(
let results = if form.search.is_empty() {
Vec::new()
} else {
- let query = form.search.clone();
- state.db_query(move |db| {
- db.search_stories(&query, 20, 0.5)
- }).await?
+ state.db().search_stories(&form.search, 20, 0.5).await?
};
let template = SearchResultsTemplate { results };
diff --git a/src/handlers/story.rs b/src/handlers/story.rs
@@ -20,49 +20,49 @@ async fn fetch_enriched_comments(
user_id: Option<i64>,
story_id: i64,
) -> Result<Vec<CommentWithMeta>, AppError> {
- state.db_query(move |db| {
- let comments = db.get_comments_for_story(story_id)?;
-
- let comment_ids: Vec<i64> = comments.iter().map(|c| c.id).collect();
- let voted_comment_ids = if let Some(uid) = user_id {
- if !comment_ids.is_empty() {
- db.get_user_votes(uid, TARGET_TYPE_COMMENT, &comment_ids)
- .unwrap_or_default()
- } else {
- vec![]
- }
+ let db = state.db();
+ let comments = db.get_comments_for_story(story_id).await?;
+
+ let comment_ids: Vec<i64> = comments.iter().map(|c| c.id).collect();
+ let voted_comment_ids = if let Some(uid) = user_id {
+ if !comment_ids.is_empty() {
+ db.get_user_votes(uid, TARGET_TYPE_COMMENT, &comment_ids)
+ .await
+ .unwrap_or_default()
} else {
vec![]
- };
-
- // Batch-fetch comment author usernames
- let unique_user_ids: Vec<i64> = {
- let mut ids: Vec<i64> = comments.iter()
- .filter_map(|c| c.created_by.parse::<i64>().ok())
- .collect();
- ids.sort_unstable();
- ids.dedup();
- ids
- };
- let author_map = db.enrich_comment_authors(&unique_user_ids).unwrap_or_default();
-
- let mut enriched = Vec::with_capacity(comments.len());
- for comment in comments {
- let uid: i64 = comment.created_by.parse().unwrap_or(0);
- let comment_author = author_map.get(&uid).cloned();
- let comment_time_ago = database::time_ago(comment.created_at);
- let comment_voted = voted_comment_ids.contains(&comment.id);
-
- enriched.push(CommentWithMeta::from_comment(
- comment,
- comment_author,
- comment_time_ago,
- comment_voted,
- ));
}
+ } else {
+ vec![]
+ };
- Ok(enriched)
- }).await
+ // Batch-fetch comment author usernames
+ let unique_user_ids: Vec<i64> = {
+ let mut ids: Vec<i64> = comments.iter()
+ .filter_map(|c| c.created_by.parse::<i64>().ok())
+ .collect();
+ ids.sort_unstable();
+ ids.dedup();
+ ids
+ };
+ let author_map = db.enrich_comment_authors(&unique_user_ids).await.unwrap_or_default();
+
+ let mut enriched = Vec::with_capacity(comments.len());
+ for comment in comments {
+ let uid: i64 = comment.created_by.parse().unwrap_or(0);
+ let comment_author = author_map.get(&uid).cloned();
+ let comment_time_ago = database::time_ago(comment.created_at);
+ let comment_voted = voted_comment_ids.contains(&comment.id);
+
+ enriched.push(CommentWithMeta::from_comment(
+ comment,
+ comment_author,
+ comment_time_ago,
+ comment_voted,
+ ));
+ }
+
+ Ok(enriched)
}
pub async fn show_story(
@@ -72,35 +72,34 @@ pub async fn show_story(
) -> Result<Html<String>, AppError> {
let user_id = state.get_user_id(&jar).await;
- let (story, tags, author_username, story_voted) = state.db_query(move |db| {
- let story = db.get_story(id)?
- .ok_or_else(|| AppError::NotFound("Story not found".to_string()))?;
+ let db = state.db();
+ let story = db.get_story(id).await?
+ .ok_or_else(|| AppError::NotFound("Story not found".to_string()))?;
- let tags = db.get_categories_for_story(id).unwrap_or_default();
+ let tags = db.get_categories_for_story(id).await.unwrap_or_default();
- let author_username = if let Some(ref created_by) = story.created_by {
- if let Ok(uid) = created_by.parse::<i64>() {
- db.get_user_by_id(uid)
- .ok()
- .flatten()
- .map(|u| u.username)
- } else {
- None
- }
+ let author_username = if let Some(ref created_by) = story.created_by {
+ if let Ok(uid) = created_by.parse::<i64>() {
+ db.get_user_by_id(uid)
+ .await
+ .ok()
+ .flatten()
+ .map(|u| u.username)
} else {
None
- };
-
- let story_voted = if let Some(uid) = user_id {
- db.get_user_votes(uid, TARGET_TYPE_STORY, &[id])
- .map(|v| v.contains(&id))
- .unwrap_or(false)
- } else {
- false
- };
+ }
+ } else {
+ None
+ };
- Ok((story, tags, author_username, story_voted))
- }).await?;
+ let story_voted = if let Some(uid) = user_id {
+ db.get_user_votes(uid, TARGET_TYPE_STORY, &[id])
+ .await
+ .map(|v| v.contains(&id))
+ .unwrap_or(false)
+ } else {
+ false
+ };
let time_ago = database::time_ago(story.published);
let story_with_meta = StoryWithMeta::from_story(story, tags, author_username, time_ago, story_voted, 0);
@@ -134,7 +133,7 @@ pub async fn show_submit(
));
}
- let all_tags = state.db_query(|db| db.get_categories(50)).await.unwrap_or_default();
+ let all_tags = state.db().get_categories(50).await.unwrap_or_default();
let submit = SubmitTemplate {
error: None,
@@ -169,7 +168,7 @@ pub async fn submit_story(
Html(r#"<meta http-equiv="refresh" content="0;url=/login">"#.to_string())
})?;
- let all_tags = state.db_query(|db| db.get_categories(50)).await.unwrap_or_default();
+ let all_tags = state.db().get_categories(50).await.unwrap_or_default();
// Validate: must have either URL or text
let url = form.url.as_ref().filter(|u| !u.trim().is_empty());
@@ -201,27 +200,22 @@ pub async fn submit_story(
let text_val = text.map(|s| s.to_string()).unwrap_or_default();
let tag_ids = form.tags.clone();
- let story = state.db_query(move |db| {
- db.create_story(
- uid,
- &title,
- url_val.as_deref(),
- &text_val,
- domain.as_deref(),
- "en",
- )
- }).await.map_err(|e| render_submit_error(&form, &all_tags, &format!("Failed to create story: {}", e)))?;
+ let story = state.db().create_story(
+ uid,
+ &title,
+ url_val.as_deref(),
+ &text_val,
+ domain.as_deref(),
+ "en",
+ ).await.map_err(|e| render_submit_error(&form, &all_tags, &format!("Failed to create story: {}", e)))?;
let story_id = story.id;
if !tag_ids.is_empty() {
- let _ = state.db_query(move |db| {
- for tag_id in tag_ids {
- if let Err(e) = db.add_tag_to_story(story_id, tag_id) {
- eprintln!("Failed to add tag {} to story {}: {}", tag_id, story_id, e);
- }
+ for tag_id in tag_ids {
+ if let Err(e) = state.db().add_tag_to_story(story_id, tag_id).await {
+ eprintln!("Failed to add tag {} to story {}: {}", tag_id, story_id, e);
}
- Ok(())
- }).await;
+ }
}
Ok(Redirect::to(&format!("/story/{}", story.id)))
@@ -254,23 +248,20 @@ pub async fn show_edit(
let user_id = state.get_user_id(&jar).await
.ok_or_else(|| AppError::Unauthorized("Login required".to_string()))?;
- let (story, all_tags, story_tags) = state.db_query(move |db| {
- let story = db.get_story(id)?
- .ok_or_else(|| AppError::NotFound("Story not found".to_string()))?;
-
- // Check ownership
- let story_owner: i64 = story.created_by.as_ref()
- .and_then(|s| s.parse().ok())
- .unwrap_or(0);
- if story_owner != user_id {
- return Err(AppError::Unauthorized("You can only edit your own stories".to_string()));
- }
+ let db = state.db();
+ let story = db.get_story(id).await?
+ .ok_or_else(|| AppError::NotFound("Story not found".to_string()))?;
- let all_tags = db.get_categories(50).unwrap_or_default();
- let story_tags = db.get_categories_for_story(id).unwrap_or_default();
+ // Check ownership
+ let story_owner: i64 = story.created_by.as_ref()
+ .and_then(|s| s.parse().ok())
+ .unwrap_or(0);
+ if story_owner != user_id {
+ return Err(AppError::Unauthorized("You can only edit your own stories".to_string()));
+ }
- Ok((story, all_tags, story_tags))
- }).await?;
+ let all_tags = db.get_categories(50).await.unwrap_or_default();
+ let story_tags = db.get_categories_for_story(id).await.unwrap_or_default();
let selected_tags: Vec<i64> = story_tags.iter().map(|t| t.id).collect();
@@ -310,9 +301,9 @@ pub async fn edit_story(
})?;
// Check ownership
- let story = state.db_query(move |db| {
- db.get_story(id)?.ok_or_else(|| AppError::NotFound("Story not found".to_string()))
- }).await.map_err(|e| render_edit_error(id, &form, &[], &format!("{}", e)))?;
+ let story = state.db().get_story(id).await
+ .and_then(|opt| opt.ok_or_else(|| AppError::NotFound("Story not found".to_string())))
+ .map_err(|e| render_edit_error(id, &form, &[], &format!("{}", e)))?;
let story_owner: i64 = story.created_by.as_ref()
.and_then(|s| s.parse().ok())
@@ -322,7 +313,7 @@ pub async fn edit_story(
}
if form.title.trim().is_empty() {
- let all_tags = state.db_query(|db| db.get_categories(50)).await.unwrap_or_default();
+ let all_tags = state.db().get_categories(50).await.unwrap_or_default();
return Err(render_edit_error(id, &form, &all_tags, "Title is required"));
}
@@ -335,18 +326,17 @@ pub async fn edit_story(
let text_val = text.map(|s| s.to_string()).unwrap_or_default();
let tag_ids = form.tags.clone();
- state.db_query(move |db| {
- db.update_story(id, &title, url_val.as_deref(), &text_val, domain.as_deref())?;
+ let db = state.db();
+ db.update_story(id, &title, url_val.as_deref(), &text_val, domain.as_deref()).await
+ .map_err(|e| render_edit_error(id, &form, &[], &format!("Failed to update: {}", e)))?;
- db.remove_all_tags_from_story(id)?;
- for tag_id in tag_ids {
- if let Err(e) = db.add_tag_to_story(id, tag_id) {
- eprintln!("Failed to add tag {} to story {}: {}", tag_id, id, e);
- }
+ db.remove_all_tags_from_story(id).await
+ .map_err(|e| render_edit_error(id, &form, &[], &format!("Failed to update: {}", e)))?;
+ for tag_id in tag_ids {
+ if let Err(e) = db.add_tag_to_story(id, tag_id).await {
+ eprintln!("Failed to add tag {} to story {}: {}", tag_id, id, e);
}
-
- Ok(())
- }).await.map_err(|e| render_edit_error(id, &form, &[], &format!("Failed to update: {}", e)))?;
+ }
Ok(Redirect::to(&format!("/story/{}", id)))
}
diff --git a/src/handlers/tag.rs b/src/handlers/tag.rs
@@ -31,60 +31,60 @@ pub async fn show_tag(
let page = query.page;
let tag_name_clone = tag_name.clone();
- let (stories, reached_end) = state.db_query(move |db| {
- // Find the category by name
- let categories = db.get_categories(100)?;
- let category = categories
- .iter()
- .find(|c| c.name.as_deref() == Some(&tag_name_clone))
- .ok_or_else(|| AppError::NotFound(format!("Tag '{}' not found", tag_name_clone)))?;
-
- let stories = db.get_stories_by_tag(category.id, PAGE_LIMIT + 1, offset)?;
-
- let reached_end = stories.len() <= PAGE_LIMIT;
- let stories: Vec<_> = stories.into_iter().take(PAGE_LIMIT).collect();
-
- let story_ids: Vec<i64> = stories.iter().map(|s| s.id).collect();
-
- let voted_ids = if let Some(uid) = user_id {
- if !story_ids.is_empty() {
- db.get_user_votes(uid, TARGET_TYPE_STORY, &story_ids)
- .unwrap_or_default()
- } else {
- vec![]
- }
+ let db = state.db();
+ // Find the category by name
+ let categories = db.get_categories(100).await?;
+ let category = categories
+ .iter()
+ .find(|c| c.name.as_deref() == Some(&tag_name_clone))
+ .ok_or_else(|| AppError::NotFound(format!("Tag '{}' not found", tag_name_clone)))?;
+
+ let all_stories = db.get_stories_by_tag(category.id, PAGE_LIMIT + 1, offset).await?;
+
+ let reached_end = all_stories.len() <= PAGE_LIMIT;
+ let stories: Vec<_> = all_stories.into_iter().take(PAGE_LIMIT).collect();
+
+ let story_ids: Vec<i64> = stories.iter().map(|s| s.id).collect();
+
+ let voted_ids = if let Some(uid) = user_id {
+ if !story_ids.is_empty() {
+ db.get_user_votes(uid, TARGET_TYPE_STORY, &story_ids)
+ .await
+ .unwrap_or_default()
} else {
vec![]
- };
-
- let meta = db.enrich_stories(&story_ids).unwrap_or_default();
-
- let base_rank = page * PAGE_LIMIT;
- let mut enriched_stories = Vec::with_capacity(stories.len());
-
- for (idx, story) in stories.into_iter().enumerate() {
- let story_meta = meta.iter().find(|m| m.story_id == story.id);
- let tags: Vec<Category> = story_meta
- .map(|m| m.tags.iter().map(|name| Category { id: 0, name: Some(name.clone()) }).collect())
- .unwrap_or_default();
- let author_username = story_meta.and_then(|m| m.author_username.clone());
-
- let time_ago = database::time_ago(story.published);
- let user_voted = voted_ids.contains(&story.id);
- let rank = base_rank + idx + 1;
-
- enriched_stories.push(StoryWithMeta::from_story(
- story,
- tags,
- author_username,
- time_ago,
- user_voted,
- rank,
- ));
}
+ } else {
+ vec![]
+ };
- Ok((enriched_stories, reached_end))
- }).await?;
+ let meta = db.enrich_stories(&story_ids).await.unwrap_or_default();
+
+ let base_rank = page * PAGE_LIMIT;
+ let mut enriched_stories = Vec::with_capacity(stories.len());
+
+ for (idx, story) in stories.into_iter().enumerate() {
+ let story_meta = meta.iter().find(|m| m.story_id == story.id);
+ let tags: Vec<Category> = story_meta
+ .map(|m| m.tags.iter().map(|name| Category { id: 0, name: Some(name.clone()) }).collect())
+ .unwrap_or_default();
+ let author_username = story_meta.and_then(|m| m.author_username.clone());
+
+ let time_ago = database::time_ago(story.published);
+ let user_voted = voted_ids.contains(&story.id);
+ let rank = base_rank + idx + 1;
+
+ enriched_stories.push(StoryWithMeta::from_story(
+ story,
+ tags,
+ author_username,
+ time_ago,
+ user_voted,
+ rank,
+ ));
+ }
+
+ let stories = enriched_stories;
let is_logged_in = user_id.is_some();
diff --git a/src/handlers/user.rs b/src/handlers/user.rs
@@ -32,72 +32,70 @@ pub async fn show_user(
let tab = query.tab.clone();
let username_clone = username.clone();
- let (profile, submissions, comments) = state.db_query(move |db| {
- // Get user by username
- let user = db.get_user_by_username(&username_clone)?
- .ok_or_else(|| AppError::NotFound(format!("User '{}' not found", username_clone)))?;
- let profile: UserProfile = user.into();
-
- let mut submissions: Vec<StoryWithMeta> = Vec::new();
- let mut comments: Vec<CommentWithMeta> = Vec::new();
-
- let user_id_num: i64 = profile.user_id.parse().unwrap_or(0);
-
- match tab.as_str() {
- "submissions" => {
- let stories = db.get_stories_by_user(user_id_num, 30, 0)?;
-
- let story_ids: Vec<i64> = stories.iter().map(|s| s.id).collect();
- let voted_ids = if let Some(uid) = current_user_id {
- if !story_ids.is_empty() {
- db.get_user_votes(uid, TARGET_TYPE_STORY, &story_ids)
- .unwrap_or_default()
- } else {
- vec![]
- }
+ let db = state.db();
+ // Get user by username
+ let user = db.get_user_by_username(&username_clone).await?
+ .ok_or_else(|| AppError::NotFound(format!("User '{}' not found", username_clone)))?;
+ let profile: UserProfile = user.into();
+
+ let mut submissions: Vec<StoryWithMeta> = Vec::new();
+ let mut comments: Vec<CommentWithMeta> = Vec::new();
+
+ let user_id_num: i64 = profile.user_id.parse().unwrap_or(0);
+
+ match tab.as_str() {
+ "submissions" => {
+ let stories = db.get_stories_by_user(user_id_num, 30, 0).await?;
+
+ let story_ids: Vec<i64> = stories.iter().map(|s| s.id).collect();
+ let voted_ids = if let Some(uid) = current_user_id {
+ if !story_ids.is_empty() {
+ db.get_user_votes(uid, TARGET_TYPE_STORY, &story_ids)
+ .await
+ .unwrap_or_default()
} else {
vec![]
- };
-
- // Batch fetch tags
- let meta = db.enrich_stories(&story_ids).unwrap_or_default();
-
- for (idx, story) in stories.into_iter().enumerate() {
- let story_meta = meta.iter().find(|m| m.story_id == story.id);
- let tags = story_meta
- .map(|m| m.tags.iter().map(|name| database::Category { id: 0, name: Some(name.clone()) }).collect())
- .unwrap_or_default();
- let time_ago = database::time_ago(story.published);
- let user_voted = voted_ids.contains(&story.id);
-
- submissions.push(StoryWithMeta::from_story(
- story,
- tags,
- Some(profile.username.clone()),
- time_ago,
- user_voted,
- idx + 1,
- ));
}
+ } else {
+ vec![]
+ };
+
+ // Batch fetch tags
+ let meta = db.enrich_stories(&story_ids).await.unwrap_or_default();
+
+ for (idx, story) in stories.into_iter().enumerate() {
+ let story_meta = meta.iter().find(|m| m.story_id == story.id);
+ let tags = story_meta
+ .map(|m| m.tags.iter().map(|name| database::Category { id: 0, name: Some(name.clone()) }).collect())
+ .unwrap_or_default();
+ let time_ago = database::time_ago(story.published);
+ let user_voted = voted_ids.contains(&story.id);
+
+ submissions.push(StoryWithMeta::from_story(
+ story,
+ tags,
+ Some(profile.username.clone()),
+ time_ago,
+ user_voted,
+ idx + 1,
+ ));
}
- "comments" => {
- let user_comments = db.get_comments_by_user(user_id_num, 30, 0)?;
-
- for comment in user_comments {
- let time_ago = database::time_ago(comment.created_at);
- comments.push(CommentWithMeta::from_comment(
- comment,
- Some(profile.username.clone()),
- time_ago,
- false,
- ));
- }
+ }
+ "comments" => {
+ let user_comments = db.get_comments_by_user(user_id_num, 30, 0).await?;
+
+ for comment in user_comments {
+ let time_ago = database::time_ago(comment.created_at);
+ comments.push(CommentWithMeta::from_comment(
+ comment,
+ Some(profile.username.clone()),
+ time_ago,
+ false,
+ ));
}
- _ => {}
}
-
- Ok((profile, submissions, comments))
- }).await?;
+ _ => {}
+ }
let is_logged_in = current_user_id.is_some();
diff --git a/src/handlers/vote.rs b/src/handlers/vote.rs
@@ -37,11 +37,9 @@ pub async fn vote_story(
let user_id = state.get_user_id(&jar).await
.ok_or_else(|| AppError::Unauthorized("Login required to vote".to_string()))?;
- let (voted, new_score) = state.db_query(move |db| {
- let voted = db.toggle_vote(user_id, TARGET_TYPE_STORY, story_id)?;
- let new_score = db.get_story_score(story_id)?;
- Ok((voted, new_score))
- }).await?;
+ let db = state.db();
+ let voted = db.toggle_vote(user_id, TARGET_TYPE_STORY, story_id).await?;
+ let new_score = db.get_story_score(story_id).await?;
render_vote_button("story", story_id, new_score, voted)
}
@@ -55,11 +53,9 @@ pub async fn vote_comment(
let user_id = state.get_user_id(&jar).await
.ok_or_else(|| AppError::Unauthorized("Login required to vote".to_string()))?;
- let (voted, new_score) = state.db_query(move |db| {
- let voted = db.toggle_vote(user_id, TARGET_TYPE_COMMENT, comment_id)?;
- let new_score = db.get_comment_score(comment_id)?;
- Ok((voted, new_score))
- }).await?;
+ let db = state.db();
+ let voted = db.toggle_vote(user_id, TARGET_TYPE_COMMENT, comment_id).await?;
+ let new_score = db.get_comment_score(comment_id).await?;
render_vote_button("comment", comment_id, new_score, voted)
}
diff --git a/src/main.rs b/src/main.rs
@@ -13,19 +13,8 @@ use config::Config;
use database::Database;
use state::AppState;
-fn main() -> Result<(), Box<dyn std::error::Error>> {
- let cores = std::thread::available_parallelism()
- .map(|n| n.get())
- .unwrap_or(4);
-
- tokio::runtime::Builder::new_multi_thread()
- .enable_all()
- .max_blocking_threads(cores)
- .build()?
- .block_on(async_main())
-}
-
-async fn async_main() -> Result<(), Box<dyn std::error::Error>> {
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
let config = Config::from_env();
eprintln!("Starting server with DEBUG={}", config.debug);
diff --git a/src/state.rs b/src/state.rs
@@ -2,49 +2,33 @@ use std::sync::Arc;
use crate::config::Config;
use crate::database::Database;
-use crate::error::AppError;
#[derive(Clone)]
pub struct AppState {
pub config: Arc<Config>,
pub db: Arc<Database>,
pub comment_tx: tokio::sync::broadcast::Sender<i64>,
- /// Limits concurrent password hashing so argon2 can't fill
- /// all blocking threads and starve DB reads.
- pub hash_semaphore: Arc<tokio::sync::Semaphore>,
}
impl AppState {
pub fn new(config: Config, db: Database) -> Self {
let (comment_tx, _) = tokio::sync::broadcast::channel(64);
- let cores = std::thread::available_parallelism()
- .map(|n| n.get())
- .unwrap_or(4);
Self {
config: Arc::new(config),
db: Arc::new(db),
comment_tx,
- hash_semaphore: Arc::new(tokio::sync::Semaphore::new((cores / 2).max(1))),
}
}
/// Look up session token in DB, return user_id if valid.
pub async fn get_user_id(&self, jar: &axum_extra::extract::cookie::CookieJar) -> Option<i64> {
let token = crate::handlers::auth::get_session_token(jar)?;
- self.db_query(move |db| db.get_session_user_id(&token))
- .await
- .ok()
- .flatten()
+ self.db.get_session_user_id(&token).await.ok().flatten()
}
- pub async fn db_query<F, T>(&self, f: F) -> Result<T, AppError>
- where
- F: FnOnce(&Database) -> Result<T, AppError> + Send + 'static,
- T: Send + 'static,
- {
- let db = self.db.clone();
- tokio::task::spawn_blocking(move || f(&db))
- .await
- .map_err(|e| AppError::Database(e.to_string()))?
+ /// Access the database. Reads/writes dispatch to dedicated thread
+ /// pools and return via tokio oneshot — fully async, no spawn_blocking.
+ pub fn db(&self) -> &Database {
+ &self.db
}
}
