实现实时双写同步并接入核心写操作

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<RwLock<AppState>>>,
+    app_state: &Arc<RwLock<AppState>>,
 ) -> 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<RwLock<AppState>>) -> 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<RwLock<AppState>>,
+) -> Result<DbSyncApplyResult, String> {
+    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<RwLock<AppState>>) -> 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<IpcResponse<DbSyncPreviewResult>, 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<RwLock<AppState>>>,
 ) -> Result<IpcResponse<DbSyncApplyResult>, 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]

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")),

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

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 {

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<RwLock<AppState>>, sender_id: Option<u32>) -> 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"))