From da412c1c76e80872a61de9597726494e83c44f49 Mon Sep 17 00:00:00 2001 From: lychang Date: Mon, 20 Jul 2026 00:50:45 +0800 Subject: [PATCH] update --- src/db/album.rs | 2 ++ src/db/connection.rs | 19 +++++++----- src/db/favorite.rs | 3 ++ src/db/image.rs | 1 + src/db/metadata.rs | 67 +++++++++++++++++++++++++++++++++++++++++- src/db/migration.rs | 3 ++ src/engine/pipeline.rs | 43 +++++++++++++++++++-------- 7 files changed, 118 insertions(+), 20 deletions(-) diff --git a/src/db/album.rs b/src/db/album.rs index 3e05928..cf78940 100644 --- a/src/db/album.rs +++ b/src/db/album.rs @@ -98,6 +98,7 @@ impl Database { // ── 写入 ── pub async fn create_album(&self, album: CreateAlbum) -> Result { + let _guard = self.write_lock.lock().await; let id = Uuid::new_v4(); let now = chrono::Utc::now().to_rfc3339(); sqlx::query( @@ -108,6 +109,7 @@ impl Database { } pub async fn delete_album(&self, album_id: Uuid) -> Result<(), sqlx::Error> { + let _guard = self.write_lock.lock().await; sqlx::query("DELETE FROM albums WHERE id = ?").bind(album_id.to_string()).execute(&self.pool).await?; Ok(()) } diff --git a/src/db/connection.rs b/src/db/connection.rs index e164da2..ca29f8e 100644 --- a/src/db/connection.rs +++ b/src/db/connection.rs @@ -1,9 +1,12 @@ use sqlx::{SqlitePool, sqlite::SqlitePoolOptions}; use crate::config::Config; +use tokio::sync::Mutex; pub struct Database { - pub pool: SqlitePool, + pub(crate) pool: SqlitePool, pub config: Config, + /// 写互斥锁：单协程串行化所有写操作，消除 WAL 锁争用 + pub(crate) write_lock: Mutex<()>, } impl Database { @@ -22,26 +25,28 @@ impl Database { } let pool = SqlitePoolOptions::new() - .max_connections(10) - .min_connections(2) + .max_connections(2) // 写串行化后无需过多连接 + .min_connections(1) .acquire_timeout(std::time::Duration::from_secs(5)) .idle_timeout(std::time::Duration::from_secs(300)) .max_lifetime(std::time::Duration::from_secs(1800)) .connect(url) .await?; - // SQLite 性能优化 + // SQLite 高并发优化 PRAGMA sqlx::query("PRAGMA journal_mode=WAL;").execute(&pool).await?; - sqlx::query("PRAGMA wal_autocheckpoint=4000;").execute(&pool).await?; sqlx::query("PRAGMA synchronous=NORMAL;").execute(&pool).await?; - sqlx::query("PRAGMA cache_size=-64000;").execute(&pool).await?; + sqlx::query("PRAGMA busy_timeout=5000;").execute(&pool).await?; // 等待 5s 避免 SQLITE_BUSY + sqlx::query("PRAGMA wal_autocheckpoint=1000;").execute(&pool).await?; // 小 WAL → 短 checkpoint + sqlx::query("PRAGMA journal_size_limit=67108864;").execute(&pool).await?; // WAL 上限 64MB + sqlx::query("PRAGMA cache_size=-64000;").execute(&pool).await?; // 64KB 页缓存 sqlx::query("PRAGMA mmap_size=268435456;").execute(&pool).await?; sqlx::query("PRAGMA temp_store=MEMORY;").execute(&pool).await?; sqlx::query("PRAGMA foreign_keys=ON;").execute(&pool).await?; sqlx::migrate!("./migrations").run(&pool).await?; - let db = Database { pool, config: config.clone() }; + let db = Database { pool, config: config.clone(), write_lock: Mutex::new(()) }; db.ensure_image_count_field().await?; Ok(db) } diff --git a/src/db/favorite.rs b/src/db/favorite.rs index 9c5efe7..67d6cc4 100644 --- a/src/db/favorite.rs +++ b/src/db/favorite.rs @@ -46,6 +46,7 @@ impl Database { } pub async fn add_favorite(&self, user_id: &str, fav: CreateFavorite) -> Result { + let _guard = self.write_lock.lock().await; let id = Uuid::new_v4(); let now = chrono::Utc::now().to_rfc3339(); sqlx::query("INSERT INTO favorites (id, user_id, image_id, album_id, created_at) VALUES (?, ?, ?, ?, ?)") @@ -55,11 +56,13 @@ impl Database { } pub async fn remove_favorite(&self, favorite_id: Uuid) -> Result<(), sqlx::Error> { + let _guard = self.write_lock.lock().await; sqlx::query("DELETE FROM favorites WHERE id = ?").bind(favorite_id.to_string()).execute(&self.pool).await?; Ok(()) } pub async fn remove_favorite_by_image(&self, user_id: &str, image_id: Uuid) -> Result<(), sqlx::Error> { + let _guard = self.write_lock.lock().await; sqlx::query("DELETE FROM favorites WHERE user_id = ? AND image_id = ?") .bind(user_id).bind(image_id.to_string()).execute(&self.pool).await?; Ok(()) diff --git a/src/db/image.rs b/src/db/image.rs index 90e1269..60d05b2 100644 --- a/src/db/image.rs +++ b/src/db/image.rs @@ -8,6 +8,7 @@ pub use crate::models::image::Image; impl Database { pub async fn add_images(&self, images: Vec) -> Result, sqlx::Error> { + let _guard = self.write_lock.lock().await; if images.is_empty() { return Ok(Vec::new()); } let now = chrono::Utc::now().to_rfc3339(); let mut ids = Vec::new(); diff --git a/src/db/metadata.rs b/src/db/metadata.rs index 458ab95..37bb747 100644 --- a/src/db/metadata.rs +++ b/src/db/metadata.rs @@ -1,6 +1,6 @@ use super::Database; use crate::models::*; -use sqlx::{Acquire, Row}; +use sqlx::{Acquire, Row, Sqlite, Transaction}; use uuid::Uuid; impl Database { @@ -157,22 +157,48 @@ impl Database { self.get_person_by_name(name).await.map(|p| p.id).unwrap_or(0) } + /// 与 insert_person 相同，但使用已有的事务 + pub async fn insert_person_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, name: &str, display_name: Option<&str>) -> i64 { + if let Ok(Some(r)) = sqlx::query("SELECT id FROM persons WHERE name = ?") + .bind(name).fetch_optional(&mut *tx).await { + return r.get::("id"); + } + sqlx::query("INSERT OR IGNORE INTO persons (name, display_name) VALUES (?, ?)") + .bind(name).bind(display_name).execute(&mut *tx).await.ok(); + if let Ok(Some(r)) = sqlx::query("SELECT id FROM persons WHERE name = ?") + .bind(name).fetch_optional(&mut *tx).await { + return r.get::("id"); + } + 0 + } + pub async fn insert_person_alias(&self, person_id: i64, alias_type: &str, alias_name: &str) { sqlx::query("INSERT OR IGNORE INTO person_aliases (person_id, alias_type, alias_name) VALUES (?, ?, ?)") .bind(person_id).bind(alias_type).bind(alias_name).execute(&self.pool).await.ok(); } + pub async fn insert_person_alias_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, person_id: i64, alias_type: &str, alias_name: &str) { + sqlx::query("INSERT OR IGNORE INTO person_aliases (person_id, alias_type, alias_name) VALUES (?, ?, ?)") + .bind(person_id).bind(alias_type).bind(alias_name).execute(&mut *tx).await.ok(); + } + pub async fn insert_person_album(&self, person_id: i64, album_id: &str) { sqlx::query("INSERT OR IGNORE INTO person_albums (person_id, album_id) VALUES (?, ?)") .bind(person_id).bind(album_id).execute(&self.pool).await.ok(); } + pub async fn insert_person_album_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, person_id: i64, album_id: &str) { + sqlx::query("INSERT OR IGNORE INTO person_albums (person_id, album_id) VALUES (?, ?)") + .bind(person_id).bind(album_id).execute(&mut *tx).await.ok(); + } + pub async fn delete_person_albums(&self, person_id: i64) { sqlx::query("DELETE FROM person_albums WHERE person_id = ?").bind(person_id).execute(&self.pool).await.ok(); } /// 彻底删除一个人物（包括别名、相册关联、自身） pub async fn delete_person_completely(&self, person_id: i64) { + let _guard = self.write_lock.lock().await; sqlx::query("DELETE FROM person_aliases WHERE person_id = ?").bind(person_id).execute(&self.pool).await.ok(); sqlx::query("DELETE FROM person_albums WHERE person_id = ?").bind(person_id).execute(&self.pool).await.ok(); sqlx::query("DELETE FROM persons WHERE id = ?").bind(person_id).execute(&self.pool).await.ok(); @@ -223,6 +249,11 @@ impl Database { .bind(person_name_original).bind(person_name_cleaned).bind(album_id).execute(&self.pool).await.ok(); } + pub async fn update_album_person_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, album_id: &str, person_name_original: &str, person_name_cleaned: &str) { + sqlx::query("UPDATE albums SET person_name_original = ?, person_name_cleaned = ?, processed_at = datetime('now') WHERE id = ?") + .bind(person_name_original).bind(person_name_cleaned).bind(album_id).execute(&mut *tx).await.ok(); + } + /// 清空该人物所有关联相册的 person 字段（当该人物被确认为非人物后调用） pub async fn update_album_person_by_person_id(&self, person_id: i64) { sqlx::query( @@ -266,6 +297,32 @@ impl Database { } } + /// 与 update_album_tags 相同，但使用已有的事务（不启动内部子事务） + pub async fn update_album_tags_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, album_id: &str, tags_json: &str, all_words_json: &str) { + sqlx::query("UPDATE albums SET tags = ?, all_words = ?, processed_at = datetime('now') WHERE id = ?") + .bind(tags_json).bind(all_words_json).bind(album_id).execute(&mut *tx).await.ok(); + + sqlx::query("DELETE FROM album_tags WHERE album_id = ?") + .bind(album_id) + .execute(&mut *tx) + .await.ok(); + + if let Ok(tags) = serde_json::from_str::>(tags_json) { + if !tags.is_empty() { + let placeholders: Vec = tags.iter().map(|_| "(?, ?)".to_string()).collect(); + let sql = format!( + "INSERT OR IGNORE INTO album_tags (album_id, tag) VALUES {}", + placeholders.join(", ") + ); + let mut q = sqlx::query(&sql); + for tag in &tags { + q = q.bind(album_id).bind(tag); + } + q.execute(&mut *tx).await.ok(); + } + } + } + pub async fn update_person_counts(&self, person_id: i64) { sqlx::query( r#"UPDATE persons SET total_albums = (SELECT COUNT(*) FROM person_albums WHERE person_id = ?), @@ -274,6 +331,14 @@ impl Database { ).bind(person_id).bind(person_id).bind(person_id).execute(&self.pool).await.ok(); } + pub async fn update_person_counts_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, person_id: i64) { + sqlx::query( + r#"UPDATE persons SET total_albums = (SELECT COUNT(*) FROM person_albums WHERE person_id = ?), + total_images = (SELECT COALESCE(SUM(a.image_count), 0) FROM albums a JOIN person_albums pa ON a.id = pa.album_id WHERE pa.person_id = ?), + updated_at = datetime('now') WHERE id = ?"# + ).bind(person_id).bind(person_id).bind(person_id).execute(&mut *tx).await.ok(); + } + /// 获取所有相册 ID 集合（用于迁移） pub async fn get_all_album_ids(&self) -> Vec { sqlx::query_scalar::<_, String>("SELECT id FROM albums").fetch_all(&self.pool).await.unwrap_or_default() diff --git a/src/db/migration.rs b/src/db/migration.rs index 69c9660..ca7e9c4 100644 --- a/src/db/migration.rs +++ b/src/db/migration.rs @@ -24,6 +24,9 @@ pub struct MigrationStats { impl Database { /// 从老数据库迁移元数据 pub async fn migrate_from_old_db(&self, old_db_path: &str) -> Result { + // 串行化写入：迁移过程中禁止其他写入操作 + let _guard = self.write_lock.lock().await; + if !std::path::Path::new(old_db_path).exists() { return Err(format!("老数据库不存在: {}", old_db_path)); } diff --git a/src/engine/pipeline.rs b/src/engine/pipeline.rs index c7067d9..6a76bd6 100644 --- a/src/engine/pipeline.rs +++ b/src/engine/pipeline.rs @@ -17,6 +17,7 @@ use crate::db::Database; use crate::engine::title_parser; use crate::engine::tag_extractor; use crate::engine::person_merger; +use sqlx::Acquire; use std::collections::HashSet; // ════════════════════════════════════════════════ @@ -201,17 +202,20 @@ pub async fn phase_detect_person_in_tags(db: &Database) -> (DirtySet, usize) { // ── 写入 person_name（直接提取，无需等待 Phase 1）── let (primary, aliases) = title_parser::split_person_name_with_aliases(person_name); let final_name = if primary.is_empty() { person_name.clone() } else { primary }; - db.update_album_person(&album.id, person_name, &final_name).await; + + // 用事务包裹所有写入，避免逐条 autocommit 导致巨慢 + let mut tx = db.pool.begin().await.expect("begin tx"); + db.update_album_person_with_tx(&mut tx, &album.id, person_name, &final_name).await; if !title_parser::is_excluded_name(&final_name) { - let pid = db.insert_person(&final_name, None).await; - db.insert_person_album(pid, &album.id).await; + let pid = db.insert_person_with_tx(&mut tx, &final_name, None).await; + db.insert_person_album_with_tx(&mut tx, pid, &album.id).await; for (at, an) in &aliases { if !an.is_empty() && an != &final_name { - db.insert_person_alias(pid, at, an).await; + db.insert_person_alias_with_tx(&mut tx, pid, at, an).await; } } - db.update_person_counts(pid).await; + db.update_person_counts_with_tx(&mut tx, pid).await; } // ── 写入清理后的标签 ── @@ -223,7 +227,9 @@ pub async fn phase_detect_person_in_tags(db: &Database) -> (DirtySet, usize) { .filter(|pair| !known_persons.contains(&pair[0].to_lowercase())) .collect(); let words_json_new = serde_json::to_string(&all_words_new).unwrap_or_else(|_| "[]".to_string()); - db.update_album_tags(&album.id, &tags_json_new, &words_json_new).await; + db.update_album_tags_with_tx(&mut tx, &album.id, &tags_json_new, &words_json_new).await; + + tx.commit().await.expect("commit tx"); if has_person { dirty.add(&album.id); } } @@ -323,18 +329,23 @@ pub async fn phase_parse_dirty(db: &Database, dirty_set: &DirtySet) -> (usize, u jieba_person }; + // 用事务包裹所有写入，避免逐条 autocommit 导致巨慢 + let mut pending_tx: Option> = None; + if let Some(name) = &best_person { let (primary, aliases) = title_parser::split_person_name_with_aliases(name); let final_name = if primary.is_empty() { name.clone() } else { primary }; - db.update_album_person(&album.id, name, &final_name).await; - let pid = db.insert_person(&final_name, None).await; - db.insert_person_album(pid, &album.id).await; + let mut tx = db.pool.begin().await.expect("begin tx"); + db.update_album_person_with_tx(&mut tx, &album.id, name, &final_name).await; + let pid = db.insert_person_with_tx(&mut tx, &final_name, None).await; + db.insert_person_album_with_tx(&mut tx, pid, &album.id).await; for (at, an) in &aliases { if !an.is_empty() && an != &final_name { - db.insert_person_alias(pid, at, an).await; + db.insert_person_alias_with_tx(&mut tx, pid, at, an).await; } } - db.update_person_counts(pid).await; + db.update_person_counts_with_tx(&mut tx, pid).await; + pending_tx = Some(tx); person_updated += 1; } @@ -354,8 +365,16 @@ pub async fn phase_parse_dirty(db: &Database, dirty_set: &DirtySet) -> (usize, u let tags_json = serde_json::to_string(&tag_names).unwrap_or_else(|_| "[]".to_string()); let all_words: Vec<[String; 2]> = unique.iter().map(|(t, tp)| [t.clone(), tp.clone()]).collect(); let words_json = serde_json::to_string(&all_words).unwrap_or_else(|_| "[]".to_string()); - db.update_album_tags(&album.id, &tags_json, &words_json).await; + let mut tx = match pending_tx.take() { + Some(t) => t, + None => db.pool.begin().await.expect("begin tx"), + }; + db.update_album_tags_with_tx(&mut tx, &album.id, &tags_json, &words_json).await; + tx.commit().await.expect("commit tx"); tag_updated += 1; + } else if let Some(tx) = pending_tx.take() { + // 只有 person 写入，没有标签写入 + tx.commit().await.expect("commit tx"); } }