From ec3929d65d638c3600770b1c25739139b8c00032 Mon Sep 17 00:00:00 2001 From: JSR Date: Thu, 19 Mar 2026 18:04:41 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E5=AE=9E=E6=97=B6=E5=8F=8C?= =?UTF-8?q?=E5=86=99=E5=90=8C=E6=AD=A5=E5=B9=B6=E6=8E=A5=E5=85=A5=E6=A0=B8?= =?UTF-8?q?=E5=BF=83=E5=86=99=E6=93=8D=E4=BD=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src-tauri/src/commands/database.rs | 282 ++++++++++++++++------------- src-tauri/src/commands/event.rs | 5 +- src-tauri/src/commands/reason.rs | 20 +- src-tauri/src/commands/student.rs | 14 +- src-tauri/src/commands/tag.rs | 16 +- 5 files changed, 199 insertions(+), 138 deletions(-) diff --git a/src-tauri/src/commands/database.rs b/src-tauri/src/commands/database.rs index b38bb3f..24c0cad 100644 --- a/src-tauri/src/commands/database.rs +++ b/src-tauri/src/commands/database.rs @@ -547,15 +547,15 @@ async fn ensure_student_tag_pair( Ok(true) } -async fn current_remote_and_local( +async fn current_remote_and_local_from_state( app_handle: &AppHandle, - state: &State<'_, Arc>>, + app_state: &Arc>, ) -> Result< Option<(sea_orm::DatabaseConnection, sea_orm::DatabaseConnection)>, String, > { let (current_conn, db_type) = { - let state_guard = state.read(); + let state_guard = app_state.read(); let db = state_guard.db.read().clone(); let settings = state_guard.settings.read(); let status_json = settings.get_value(SettingsKey::PgConnectionStatus); @@ -588,6 +588,158 @@ async fn current_remote_and_local( Ok(Some((local_conn, remote_conn))) } +pub async fn realtime_dual_write_sync(app_state: &Arc>) -> Result<(), String> { + let app_handle = { + let state_guard = app_state.read(); + state_guard.app_handle.clone() + }; + + let can_sync = current_remote_and_local_from_state(&app_handle, app_state) + .await? + .is_some(); + if !can_sync { + return Ok(()); + } + + let result = db_sync_apply_internal( + ConflictStrategy::KeepRemote, + app_handle, + app_state.clone(), + ) + .await?; + if !result.success { + return Err(result + .message + .unwrap_or_else(|| "实时双写同步失败".to_string())); + } + Ok(()) +} + +async fn db_sync_apply_internal( + strategy: ConflictStrategy, + app_handle: AppHandle, + app_state: Arc>, +) -> Result { + let Some((local_conn, remote_conn)) = + current_remote_and_local_from_state(&app_handle, &app_state).await? + else { + return Ok(DbSyncApplyResult { + success: false, + synced_records: 0, + resolved_conflicts: 0, + message: Some("当前不在 PostgreSQL 远程模式，无法执行同步".to_string()), + }); + }; + + let local_students = load_students(&local_conn).await?; + let remote_students = load_students(&remote_conn).await?; + let local_reasons = load_reasons(&local_conn).await?; + let remote_reasons = load_reasons(&remote_conn).await?; + let local_tags = load_tags(&local_conn).await?; + let remote_tags = load_tags(&remote_conn).await?; + let local_events = load_events(&local_conn).await?; + let remote_events = load_events(&remote_conn).await?; + let local_pairs = load_student_tag_pairs(&local_conn).await?; + let remote_pairs = load_student_tag_pairs(&remote_conn).await?; + + let (preferred, target) = if strategy == ConflictStrategy::KeepLocal { + (&local_conn, &remote_conn) + } else { + (&remote_conn, &local_conn) + }; + + let mut synced_records = 0usize; + let mut resolved_conflicts = 0usize; + + for student in local_students.values() { + if upsert_student(&remote_conn, student).await? { + synced_records += 1; + } + } + for student in remote_students.values() { + if upsert_student(&local_conn, student).await? { + synced_records += 1; + } + } + for reason in local_reasons.values() { + if upsert_reason(&remote_conn, reason).await? { + synced_records += 1; + } + } + for reason in remote_reasons.values() { + if upsert_reason(&local_conn, reason).await? { + synced_records += 1; + } + } + for tag in local_tags.values() { + if upsert_tag(&remote_conn, tag).await? { + synced_records += 1; + } + } + for tag in remote_tags.values() { + if upsert_tag(&local_conn, tag).await? { + synced_records += 1; + } + } + for event in local_events.values() { + if upsert_event(&remote_conn, event).await? { + synced_records += 1; + } + } + for event in remote_events.values() { + if upsert_event(&local_conn, event).await? { + synced_records += 1; + } + } + + for pair in local_pairs.union(&remote_pairs) { + if ensure_student_tag_pair(&local_conn, pair).await? { + synced_records += 1; + } + if ensure_student_tag_pair(&remote_conn, pair).await? { + synced_records += 1; + } + } + + let preferred_students = load_students(preferred).await?; + for student in preferred_students.values() { + if upsert_student(target, student).await? { + resolved_conflicts += 1; + } + } + let preferred_reasons = load_reasons(preferred).await?; + for reason in preferred_reasons.values() { + if upsert_reason(target, reason).await? { + resolved_conflicts += 1; + } + } + let preferred_tags = load_tags(preferred).await?; + for tag in preferred_tags.values() { + if upsert_tag(target, tag).await? { + resolved_conflicts += 1; + } + } + let preferred_events = load_events(preferred).await?; + for event in preferred_events.values() { + if upsert_event(target, event).await? { + resolved_conflicts += 1; + } + } + let preferred_pairs = load_student_tag_pairs(preferred).await?; + for pair in &preferred_pairs { + if ensure_student_tag_pair(target, pair).await? { + resolved_conflicts += 1; + } + } + + Ok(DbSyncApplyResult { + success: true, + synced_records, + resolved_conflicts, + message: Some("本地与远程同步完成".to_string()), + }) +} + fn check_admin_permission(state: &Arc>) -> Result<(), String> { let state_guard = state.read(); let mut permissions = state_guard.permissions.write(); @@ -795,7 +947,10 @@ pub async fn db_sync_preview( ) -> Result, String> { check_admin_permission(&state)?; - let Some((local_conn, remote_conn)) = current_remote_and_local(&app_handle, &state).await? else { + let app_state = state.inner().clone(); + let Some((local_conn, remote_conn)) = + current_remote_and_local_from_state(&app_handle, &app_state).await? + else { return Ok(IpcResponse::success(DbSyncPreviewResult { can_sync: false, need_sync: false, @@ -906,123 +1061,8 @@ pub async fn db_sync_apply( state: State<'_, Arc>>, ) -> Result, String> { check_admin_permission(&state)?; - - let Some((local_conn, remote_conn)) = current_remote_and_local(&app_handle, &state).await? else { - return Ok(IpcResponse::success(DbSyncApplyResult { - success: false, - synced_records: 0, - resolved_conflicts: 0, - message: Some("当前不在 PostgreSQL 远程模式，无法执行同步".to_string()), - })); - }; - - let local_students = load_students(&local_conn).await?; - let remote_students = load_students(&remote_conn).await?; - let local_reasons = load_reasons(&local_conn).await?; - let remote_reasons = load_reasons(&remote_conn).await?; - let local_tags = load_tags(&local_conn).await?; - let remote_tags = load_tags(&remote_conn).await?; - let local_events = load_events(&local_conn).await?; - let remote_events = load_events(&remote_conn).await?; - let local_pairs = load_student_tag_pairs(&local_conn).await?; - let remote_pairs = load_student_tag_pairs(&remote_conn).await?; - - let (preferred, target) = if strategy == ConflictStrategy::KeepLocal { - (&local_conn, &remote_conn) - } else { - (&remote_conn, &local_conn) - }; - - let mut synced_records = 0usize; - let mut resolved_conflicts = 0usize; - - for student in local_students.values() { - if upsert_student(&remote_conn, student).await? { - synced_records += 1; - } - } - for student in remote_students.values() { - if upsert_student(&local_conn, student).await? { - synced_records += 1; - } - } - for reason in local_reasons.values() { - if upsert_reason(&remote_conn, reason).await? { - synced_records += 1; - } - } - for reason in remote_reasons.values() { - if upsert_reason(&local_conn, reason).await? { - synced_records += 1; - } - } - for tag in local_tags.values() { - if upsert_tag(&remote_conn, tag).await? { - synced_records += 1; - } - } - for tag in remote_tags.values() { - if upsert_tag(&local_conn, tag).await? { - synced_records += 1; - } - } - for event in local_events.values() { - if upsert_event(&remote_conn, event).await? { - synced_records += 1; - } - } - for event in remote_events.values() { - if upsert_event(&local_conn, event).await? { - synced_records += 1; - } - } - - for pair in local_pairs.union(&remote_pairs) { - if ensure_student_tag_pair(&local_conn, pair).await? { - synced_records += 1; - } - if ensure_student_tag_pair(&remote_conn, pair).await? { - synced_records += 1; - } - } - - let preferred_students = load_students(preferred).await?; - for student in preferred_students.values() { - if upsert_student(target, student).await? { - resolved_conflicts += 1; - } - } - let preferred_reasons = load_reasons(preferred).await?; - for reason in preferred_reasons.values() { - if upsert_reason(target, reason).await? { - resolved_conflicts += 1; - } - } - let preferred_tags = load_tags(preferred).await?; - for tag in preferred_tags.values() { - if upsert_tag(target, tag).await? { - resolved_conflicts += 1; - } - } - let preferred_events = load_events(preferred).await?; - for event in preferred_events.values() { - if upsert_event(target, event).await? { - resolved_conflicts += 1; - } - } - let preferred_pairs = load_student_tag_pairs(preferred).await?; - for pair in &preferred_pairs { - if ensure_student_tag_pair(target, pair).await? { - resolved_conflicts += 1; - } - } - - Ok(IpcResponse::success(DbSyncApplyResult { - success: true, - synced_records, - resolved_conflicts, - message: Some("本地与远程同步完成".to_string()), - })) + let result = db_sync_apply_internal(strategy, app_handle, state.inner().clone()).await?; + Ok(IpcResponse::success(result)) } #[tauri::command] diff --git a/src-tauri/src/commands/event.rs b/src-tauri/src/commands/event.rs index b095006..2a4c0f5 100644 --- a/src-tauri/src/commands/event.rs +++ b/src-tauri/src/commands/event.rs @@ -13,6 +13,7 @@ use crate::db::entities::{score_events, students}; use crate::services::PermissionLevel; use crate::state::AppState; +use super::database::realtime_dual_write_sync; use super::response::IpcResponse; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -172,7 +173,7 @@ pub async fn event_create( active.update(&txn).await.map_err(|e| e.to_string())?; txn.commit().await.map_err(|e| e.to_string())?; - + realtime_dual_write_sync(state.inner()).await?; Ok(IpcResponse::success(inserted.id)) } Ok(None) => Ok(IpcResponse::error("Student not found")), @@ -233,7 +234,7 @@ pub async fn event_delete( .map_err(|e| e.to_string())?; txn.commit().await.map_err(|e| e.to_string())?; - + realtime_dual_write_sync(state.inner()).await?; Ok(IpcResponse::success_empty()) } Ok(None) => Ok(IpcResponse::error("Event not found")), diff --git a/src-tauri/src/commands/reason.rs b/src-tauri/src/commands/reason.rs index fb9a84e..9748f5e 100644 --- a/src-tauri/src/commands/reason.rs +++ b/src-tauri/src/commands/reason.rs @@ -8,6 +8,7 @@ use crate::db::entities::reasons; use crate::services::PermissionLevel; use crate::state::AppState; +use super::database::realtime_dual_write_sync; use super::response::IpcResponse; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -116,11 +117,14 @@ pub async fn reason_create( updated_at: Set(now), }; - match new_reason.insert(conn).await { - Ok(inserted) => Ok(IpcResponse::success(inserted.id)), - Err(e) => Ok(IpcResponse::error(&format!( - "Failed to create reason: {}", - e + match new_reason.insert(conn).await { + Ok(inserted) => { + realtime_dual_write_sync(state.inner()).await?; + Ok(IpcResponse::success(inserted.id)) + } + Err(e) => Ok(IpcResponse::error(&format!( + "Failed to create reason: {}", + e ))), } } else { @@ -165,7 +169,10 @@ pub async fn reason_update( } match active.update(conn).await { - Ok(_) => Ok(IpcResponse::success_empty()), + Ok(_) => { + realtime_dual_write_sync(state.inner()).await?; + Ok(IpcResponse::success_empty()) + } Err(e) => Ok(IpcResponse::error(&format!( "Failed to update reason: {}", e @@ -205,6 +212,7 @@ pub async fn reason_delete( if result.rows_affected == 0 { Ok(IpcResponse::error("记录不存在")) } else { + realtime_dual_write_sync(state.inner()).await?; Ok(IpcResponse::success(DeleteResult { changes: result.rows_affected as i32, })) diff --git a/src-tauri/src/commands/student.rs b/src-tauri/src/commands/student.rs index efc6223..8222d3d 100644 --- a/src-tauri/src/commands/student.rs +++ b/src-tauri/src/commands/student.rs @@ -11,6 +11,7 @@ use crate::models::{StudentUpdate, StudentWithTags}; use crate::services::PermissionLevel; use crate::state::AppState; +use super::database::realtime_dual_write_sync; use super::response::{ImportResult, IpcResponse}; #[derive(Deserialize)] @@ -126,7 +127,10 @@ pub async fn student_create( }; match new_student.insert(conn).await { - Ok(inserted) => Ok(IpcResponse::success(inserted.id)), + Ok(inserted) => { + realtime_dual_write_sync(state.inner()).await?; + Ok(IpcResponse::success(inserted.id)) + } Err(e) => Ok(IpcResponse::error(&format!( "Failed to create student: {}", e @@ -181,7 +185,10 @@ pub async fn student_update( } match active.update(conn).await { - Ok(_) => Ok(IpcResponse::success_empty()), + Ok(_) => { + realtime_dual_write_sync(state.inner()).await?; + Ok(IpcResponse::success_empty()) + } Err(e) => Ok(IpcResponse::error(&format!( "Failed to update student: {}", e @@ -230,7 +237,7 @@ pub async fn student_delete( .map_err(|e| e.to_string())?; txn.commit().await.map_err(|e| e.to_string())?; - + realtime_dual_write_sync(state.inner()).await?; Ok(IpcResponse::success_empty()) } Ok(None) => Ok(IpcResponse::error("Student not found")), @@ -314,6 +321,7 @@ pub async fn student_import_from_xlsx( } txn.commit().await.map_err(|e| e.to_string())?; + realtime_dual_write_sync(state.inner()).await?; } Ok(IpcResponse::success(ImportResult { diff --git a/src-tauri/src/commands/tag.rs b/src-tauri/src/commands/tag.rs index bfbbea2..9c61083 100644 --- a/src-tauri/src/commands/tag.rs +++ b/src-tauri/src/commands/tag.rs @@ -9,6 +9,7 @@ use crate::db::entities::{student_tags, tags}; use crate::services::PermissionLevel; use crate::state::AppState; +use super::database::realtime_dual_write_sync; use super::response::{IpcResponse, TagResponse}; fn check_admin_permission(state: &Arc>, sender_id: Option) -> bool { @@ -159,10 +160,13 @@ pub async fn tags_create( }; match new_tag.insert(conn).await { - Ok(inserted) => Ok(IpcResponse::success(TagResponse { - id: inserted.id, - name: inserted.name, - })), + Ok(inserted) => { + realtime_dual_write_sync(state.inner()).await?; + Ok(IpcResponse::success(TagResponse { + id: inserted.id, + name: inserted.name, + })) + } Err(e) => Ok(IpcResponse::error(&format!("Failed to create tag: {}", e))), } } @@ -210,7 +214,7 @@ pub async fn tags_delete( .map_err(|e| e.to_string())?; txn.commit().await.map_err(|e| e.to_string())?; - + realtime_dual_write_sync(state.inner()).await?; Ok(IpcResponse::success_empty()) } Ok(None) => Ok(IpcResponse::error("Tag not found")), @@ -264,7 +268,7 @@ pub async fn tags_update_student_tags( } txn.commit().await.map_err(|e| e.to_string())?; - + realtime_dual_write_sync(state.inner()).await?; Ok(IpcResponse::success_empty()) } else { Ok(IpcResponse::error("Database not connected"))