diff --git a/src-tauri/src/cmd/profile.rs b/src-tauri/src/cmd/profile.rs index 4c926d2a7..6ed54f332 100644 --- a/src-tauri/src/cmd/profile.rs +++ b/src-tauri/src/cmd/profile.rs @@ -213,7 +213,7 @@ async fn restore_previous_profile(prev_profile: &String) -> CmdResult<()> { async fn handle_success(current_value: Option<&String>) -> CmdResult { Config::profiles().await.apply(); - profiles::activate_selected_nodes().await.stringify_err()?; + profiles::activate_selected_nodes().stringify_err()?; handle::Handle::refresh_clash(); if let Err(e) = Tray::global().update_tooltip().await { diff --git a/src-tauri/src/config/prfitem.rs b/src-tauri/src/config/prfitem.rs index 6b992ba1f..5901387b7 100644 --- a/src-tauri/src/config/prfitem.rs +++ b/src-tauri/src/config/prfitem.rs @@ -63,7 +63,7 @@ pub struct PrfItem { pub file_data: Option, } -#[derive(Default, Debug, Clone, Deserialize, Serialize)] +#[derive(Default, Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] pub struct PrfSelected { pub name: Option, pub now: Option, diff --git a/src-tauri/src/config/profiles.rs b/src-tauri/src/config/profiles.rs index 0c6dfaff3..16374008a 100644 --- a/src-tauri/src/config/profiles.rs +++ b/src-tauri/src/config/profiles.rs @@ -1,4 +1,7 @@ -use super::{PrfOption, prfitem::PrfItem}; +use super::{ + PrfOption, + prfitem::{PrfItem, PrfSelected}, +}; use crate::{ core::handle, utils::{ @@ -12,7 +15,15 @@ use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use serde_yaml_ng::Mapping; use smartstring::alias::String; -use std::{collections::HashSet, sync::LazyLock, time::Duration}; +use std::{ + collections::HashSet, + sync::{ + LazyLock, + atomic::{AtomicU64, Ordering}, + }, + time::Duration, +}; +use tauri_plugin_mihomo::models::{Proxies, ProxyType}; use tokio::{fs, task::JoinHandle}; /// Regex to check profile file names, eg. @@ -29,6 +40,12 @@ static REGEX_PROFILE_FILE: LazyLock = // activate selected nodes task handle static ACTIVATE_SELECTED_TASK: LazyLock>>> = LazyLock::new(|| Mutex::new(None)); +static ACTIVATE_SELECTED_GENERATION: AtomicU64 = AtomicU64::new(0); + +// The plugin already limits the request/response phase to 5 seconds. This outer timeout also covers +// lock acquisition, connection-pool waiting, and local-socket connection establishment. +const MIHOMO_OPERATION_TIMEOUT: Duration = Duration::from_secs(10); +const SELECTED_NODES_RECHECK_DELAY: Duration = Duration::from_secs(1); /// Define the `profiles.yaml` schema #[derive(Default, Debug, Clone, Deserialize, Serialize)] @@ -592,89 +609,446 @@ pub async fn profiles_draft_update_item_safe(index: &String, item: &mut PrfItem) .await } -pub async fn activate_selected_nodes() -> Result<()> { - logging!(info, Type::Config, "starting activating selected nodes"); - let value = ACTIVATE_SELECTED_TASK.lock().take(); - if let Some(handle) = value { - logging!(info, Type::Config, "aborting previous worker"); - handle.abort(); - } - let profiles = Config::profiles().await.latest_arc(); - let Some(current) = profiles.get_current() else { - bail!("no current profile running"); - }; - let profile = profiles - .get_item(current) - .context("failed to get current profile")? - .clone(); +#[derive(Debug, PartialEq, Eq)] +struct SelectedNodesPlan { + selected: Vec, + activations: Vec<(String, String)>, + repaired_count: usize, +} - let handle = tokio::spawn(async move { - let mihomo = handle::Handle::mihomo().await; - // check mihomo is running - let mut is_mihomo_ready = false; - for _ in 0..10 { - if mihomo.get_version().await.is_ok() { - logging!(debug, Type::Config, "check mihomo api success"); - is_mihomo_ready = true; - break; +fn node_is_available(available_nodes: &[std::string::String], node: &str) -> bool { + available_nodes.iter().any(|available| available == node) +} + +fn selected_nodes_need_confirmation(selected: &[PrfSelected], proxies: &Proxies) -> bool { + selected.iter().any(|selected_item| { + let (Some(group_name), Some(node)) = (&selected_item.name, &selected_item.now) else { + return false; + }; + let Some(group) = proxies.proxies.get(group_name.as_str()) else { + return true; + }; + let Some(available_nodes) = group.all.as_deref().filter(|nodes| !nodes.is_empty()) else { + return true; + }; + !node_is_available(available_nodes, node) + }) +} + +fn reconcile_selected_nodes( + selected: &[PrfSelected], + previous: Option<&Proxies>, + proxies: &Proxies, +) -> SelectedNodesPlan { + let mut plan = SelectedNodesPlan { + selected: Vec::with_capacity(selected.len()), + activations: Vec::new(), + repaired_count: 0, + }; + let mut seen_groups = HashSet::new(); + let mut unique_selected = selected + .iter() + .rev() + .filter(|item| item.name.as_ref().is_some_and(|name| seen_groups.insert(name.clone()))) + .collect::>(); + unique_selected.reverse(); + plan.repaired_count += selected.len() - unique_selected.len(); + + for selected_item in unique_selected { + let (Some(group_name), Some(node)) = (&selected_item.name, &selected_item.now) else { + plan.repaired_count += 1; + continue; + }; + let Some(group) = proxies.proxies.get(group_name.as_str()) else { + if previous.is_some_and(|snapshot| !snapshot.proxies.contains_key(group_name.as_str())) { + plan.repaired_count += 1; + } else { + plan.selected.push(selected_item.clone()); } - tokio::time::sleep(Duration::from_secs(1)).await; + continue; + }; + let Some(available_nodes) = group.all.as_deref().filter(|nodes| !nodes.is_empty()) else { + // Provider-backed groups can be temporarily incomplete immediately after a reload. + plan.selected.push(selected_item.clone()); + continue; + }; + let is_selectable_group = matches!( + &group.proxy_type, + ProxyType::Selector | ProxyType::URLTest | ProxyType::Fallback | ProxyType::LoadBalance + ); + if !is_selectable_group { + let preferred_node = group + .now + .as_deref() + .filter(|current| node_is_available(available_nodes, current)) + .or_else(|| node_is_available(available_nodes, node).then_some(node.as_str())); + if let Some(preferred_node) = preferred_node { + if preferred_node != node.as_str() { + plan.repaired_count += 1; + } + plan.selected.push(PrfSelected { + name: Some(group_name.clone()), + now: Some(preferred_node.into()), + }); + } else { + plan.repaired_count += 1; + } + continue; } - if !is_mihomo_ready { - logging!( - error, + if node_is_available(available_nodes, node) { + plan.selected.push(selected_item.clone()); + if group.now.as_deref() != Some(node.as_str()) { + plan.activations.push((group_name.clone(), node.clone())); + } + continue; + } + + let missing_was_confirmed = previous + .and_then(|snapshot| snapshot.proxies.get(group_name.as_str())) + .and_then(|group| group.all.as_deref()) + .filter(|nodes| !nodes.is_empty()) + .is_some_and(|nodes| !node_is_available(nodes, node)); + if !missing_was_confirmed { + plan.selected.push(selected_item.clone()); + continue; + } + + plan.repaired_count += 1; + if let Some(current_node) = group + .now + .as_deref() + .filter(|current| node_is_available(available_nodes, current)) + { + plan.selected.push(PrfSelected { + name: Some(group_name.clone()), + now: Some(current_node.into()), + }); + } + } + + plan +} + +fn is_activation_current(generation: u64) -> bool { + ACTIVATE_SELECTED_GENERATION.load(Ordering::Acquire) == generation +} + +async fn fetch_proxies_with_timeout() -> Result { + tokio::time::timeout(MIHOMO_OPERATION_TIMEOUT, async { + loop { + match handle::Handle::mihomo().await.get_proxies().await { + Ok(proxies) => return proxies, + Err(err) => { + logging!(debug, Type::Config, "mihomo proxies are not ready yet: {err}"); + tokio::time::sleep(Duration::from_millis(500)).await; + } + } + } + }) + .await + .context("timed out while waiting for mihomo proxies") +} + +async fn select_node_with_timeout(group_name: &String, node: &String) -> Result<()> { + tokio::time::timeout(MIHOMO_OPERATION_TIMEOUT, async { + handle::Handle::mihomo() + .await + .select_node_for_group(group_name, node) + .await + }) + .await + .with_context(|| format!("timed out while selecting node [{node}] for group [{group_name}]"))? + .with_context(|| format!("failed to select node [{node}] for group [{group_name}]")) +} + +async fn persist_reconciled_selected( + profile_uid: &String, + original_selected: &[PrfSelected], + selected: Vec, + generation: u64, +) -> Result<()> { + if !is_activation_current(generation) { + return Ok(()); + } + + let profiles = Config::profiles().await; + let profile_uid = profile_uid.clone(); + let original_selected = original_selected.to_vec(); + let updated = profiles + .with_data_modify(move |mut profiles| async move { + if !is_activation_current(generation) || profiles.current.as_ref() != Some(&profile_uid) { + return Ok((profiles, false)); + } + + let profile = profiles + .items + .as_mut() + .and_then(|items| items.iter_mut().find(|item| item.uid.as_ref() == Some(&profile_uid))) + .with_context(|| format!("failed to find the profile item \"uid:{profile_uid}\""))?; + if profile.selected.as_deref().unwrap_or(&[]) != original_selected.as_slice() { + return Ok((profiles, false)); + } + + profile.selected = (!selected.is_empty()).then_some(selected); + profiles.save_file().await?; + Ok((profiles, true)) + }) + .await?; + + if updated { + handle::Handle::refresh_profiles(); + } + Ok(()) +} + +async fn activate_selected_nodes_worker( + profile_uid: String, + selected: Vec, + generation: u64, +) -> Result<()> { + let first_snapshot = fetch_proxies_with_timeout().await?; + if !is_activation_current(generation) { + return Ok(()); + } + + let (previous_snapshot, current_snapshot) = if selected_nodes_need_confirmation(&selected, &first_snapshot) { + tokio::time::sleep(SELECTED_NODES_RECHECK_DELAY).await; + if !is_activation_current(generation) { + return Ok(()); + } + let current_snapshot = fetch_proxies_with_timeout().await?; + (Some(first_snapshot), current_snapshot) + } else { + (None, first_snapshot) + }; + if !is_activation_current(generation) { + return Ok(()); + } + + let plan = reconcile_selected_nodes(&selected, previous_snapshot.as_ref(), ¤t_snapshot); + logging!(debug, Type::Config, "selected nodes activation plan: {plan:?}"); + + for (group_name, node) in &plan.activations { + if !is_activation_current(generation) { + return Ok(()); + } + match select_node_with_timeout(group_name, node).await { + Ok(()) => logging!( + info, Type::Config, - "check that the mihomo api reaches the maximum number of retries, maybe mihomo core is not running" - ); + "Selected node for proxy: {group_name}, node: {node}" + ), + Err(err) => logging!(error, Type::Config, "{err:#}"), + } + } + + if plan.repaired_count > 0 && is_activation_current(generation) { + logging!( + info, + Type::Config, + "repairing {} invalid selected node record(s) for profile {profile_uid}", + plan.repaired_count + ); + persist_reconciled_selected(&profile_uid, &selected, plan.selected, generation).await?; + } + + if is_activation_current(generation) { + handle::Handle::refresh_clash(); + } + Ok(()) +} + +pub fn activate_selected_nodes() -> Result<()> { + logging!(info, Type::Config, "starting activating selected nodes"); + let mut active_task = ACTIVATE_SELECTED_TASK.lock(); + let generation = ACTIVATE_SELECTED_GENERATION.fetch_add(1, Ordering::AcqRel) + 1; + let previous_task = active_task.take(); + + let handle = tokio::spawn(async move { + if let Some(previous_task) = previous_task { + let _ = previous_task.await; + } + if !is_activation_current(generation) { return; } - if let Some(selected) = profile.selected.as_ref() { - logging!(debug, Type::Config, "selected nodes: {selected:?}"); - for selected_item in selected { - let mut retry = 10; - if let Some(group_name) = selected_item.name.as_ref() - && let Some(node) = selected_item.now.as_ref() - { - while retry >= 0 { - logging!(debug, Type::Config, "check node[{node}] exists"); - if mihomo.get_proxy_by_name(node).await.is_ok() { - logging!(debug, Type::Config, "node[{node}] exists"); - break; - } - retry -= 1; - tokio::time::sleep(Duration::from_secs(1)).await; - } - if retry < 0 { - logging!( - error, - Type::Config, - "Failed to select node for proxy: {group_name}, node: {node}, because the node [{node}] does not exist" - ); - continue; - } - if mihomo.select_node_for_group(group_name, node).await.is_err() { - logging!( - error, - Type::Config, - "Failed to select node for proxy: {group_name}, node: {node}" - ); - } else { - logging!( - info, - Type::Config, - "Selected node for proxy: {group_name}, node: {node}" - ); - } - } + let result = async { + let profiles = Config::profiles().await.latest_arc(); + let current = profiles.get_current().context("no current profile running")?.clone(); + let selected = profiles + .get_item(¤t) + .context("failed to get current profile")? + .selected + .clone() + .unwrap_or_default(); + + if selected.is_empty() { + return Ok(()); } - // refresh clash - handle::Handle::refresh_clash(); + activate_selected_nodes_worker(current, selected, generation).await + } + .await; + + if is_activation_current(generation) { + if let Err(err) = result { + logging!(error, Type::Config, "failed to activate selected nodes: {err:#}"); + } + logging!(info, Type::Config, "activating selected nodes done!"); } - logging!(info, Type::Config, "activating selected nodes done!"); - *ACTIVATE_SELECTED_TASK.lock() = None; }); - *ACTIVATE_SELECTED_TASK.lock() = Some(handle); + *active_task = Some(handle); + drop(active_task); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + use tauri_plugin_mihomo::models::Proxy; + + fn selected(group: &str, node: &str) -> PrfSelected { + PrfSelected { + name: Some(group.into()), + now: Some(node.into()), + } + } + + fn proxies(groups: Vec<(&str, &[&str], Option<&str>)>) -> Proxies { + Proxies { + proxies: groups + .into_iter() + .map(|(name, all, now)| { + ( + name.to_owned(), + Proxy { + name: name.to_owned(), + all: Some(all.iter().map(|node| (*node).to_owned()).collect()), + now: now.map(str::to_owned), + proxy_type: ProxyType::Selector, + ..Proxy::default() + }, + ) + }) + .collect::>(), + } + } + + #[test] + fn keeps_valid_selection_and_activates_when_needed() { + let saved = vec![selected("group", "saved")]; + let plan = reconcile_selected_nodes( + &saved, + None, + &proxies(vec![("group", &["current", "saved"], Some("current"))]), + ); + + assert_eq!(plan.selected, saved); + assert_eq!(plan.activations, vec![("group".into(), "saved".into())]); + assert_eq!(plan.repaired_count, 0); + } + + #[test] + fn replaces_missing_node_with_valid_current_node() { + let snapshot = proxies(vec![("group", &["current"], Some("current"))]); + let plan = reconcile_selected_nodes(&[selected("group", "renamed-node")], Some(&snapshot), &snapshot); + + assert_eq!(plan.selected, vec![selected("group", "current")]); + assert!(plan.activations.is_empty()); + assert_eq!(plan.repaired_count, 1); + } + + #[test] + fn validates_membership_in_group_not_global_existence() { + let snapshot = proxies(vec![ + ("group", &["current"], Some("current")), + ("other-node", &[], None), + ]); + let plan = reconcile_selected_nodes(&[selected("group", "other-node")], Some(&snapshot), &snapshot); + + assert_eq!(plan.selected, vec![selected("group", "current")]); + assert!(plan.activations.is_empty()); + assert_eq!(plan.repaired_count, 1); + } + + #[test] + fn does_not_activate_non_selectable_groups() { + let snapshot = Proxies { + proxies: HashMap::from([( + "group".to_owned(), + Proxy { + name: "group".to_owned(), + all: Some(vec!["current".to_owned(), "saved".to_owned()]), + now: Some("current".to_owned()), + proxy_type: ProxyType::Direct, + ..Proxy::default() + }, + )]), + }; + + let plan = reconcile_selected_nodes(&[selected("group", "saved")], None, &snapshot); + + assert_eq!(plan.selected, vec![selected("group", "current")]); + assert!(plan.activations.is_empty()); + assert_eq!(plan.repaired_count, 1); + } + + #[test] + fn removes_selection_when_group_or_fallback_is_invalid() { + let snapshot = proxies(vec![("group", &["valid"], Some("invalid-current"))]); + let plan = reconcile_selected_nodes( + &[ + selected("missing-group", "node"), + selected("group", "missing-node"), + PrfSelected::default(), + ], + Some(&snapshot), + &snapshot, + ); + + assert!(plan.selected.is_empty()); + assert!(plan.activations.is_empty()); + assert_eq!(plan.repaired_count, 3); + } + + #[test] + fn preserves_selection_until_missing_node_is_confirmed() { + let saved = vec![selected("group", "saved")]; + let incomplete = proxies(vec![("group", &[], None)]); + let complete = proxies(vec![("group", &["current"], Some("current"))]); + + let incomplete_plan = reconcile_selected_nodes(&saved, None, &incomplete); + let one_snapshot_plan = reconcile_selected_nodes(&saved, None, &complete); + + assert_eq!(incomplete_plan.selected, saved); + assert_eq!(incomplete_plan.repaired_count, 0); + assert_eq!(one_snapshot_plan.selected, saved); + assert_eq!(one_snapshot_plan.repaired_count, 0); + } + + #[test] + fn recovers_when_group_appears_in_second_snapshot() { + let saved = vec![selected("group", "saved")]; + let incomplete = Proxies::default(); + let complete = proxies(vec![("group", &["current", "saved"], Some("current"))]); + + let plan = reconcile_selected_nodes(&saved, Some(&incomplete), &complete); + + assert_eq!(plan.selected, saved); + assert_eq!(plan.activations, vec![("group".into(), "saved".into())]); + assert_eq!(plan.repaired_count, 0); + } + + #[test] + fn keeps_last_selection_for_duplicate_group_entries() { + let saved = vec![selected("group", "old"), selected("group", "new")]; + let snapshot = proxies(vec![("group", &["old", "new"], Some("old"))]); + + let plan = reconcile_selected_nodes(&saved, None, &snapshot); + + assert_eq!(plan.selected, vec![selected("group", "new")]); + assert_eq!(plan.activations, vec![("group".into(), "new".into())]); + assert_eq!(plan.repaired_count, 1); + } +} diff --git a/src-tauri/src/core/handle.rs b/src-tauri/src/core/handle.rs index b94e711d4..bb6d7d80c 100644 --- a/src-tauri/src/core/handle.rs +++ b/src-tauri/src/core/handle.rs @@ -44,6 +44,10 @@ impl Handle { Self::send_event(FrontendEvent::RefreshVerge); } + pub fn refresh_profiles() { + Self::send_event(FrontendEvent::RefreshProfiles); + } + pub fn notify_profile_changed(profile_id: &String) { Self::send_event(FrontendEvent::ProfileChanged { current_profile_id: profile_id, diff --git a/src-tauri/src/core/notification.rs b/src-tauri/src/core/notification.rs index 782835277..a3d2edcbf 100644 --- a/src-tauri/src/core/notification.rs +++ b/src-tauri/src/core/notification.rs @@ -9,6 +9,7 @@ use tauri::{Emitter as _, WebviewWindow}; pub enum FrontendEvent<'a> { RefreshClash, RefreshVerge, + RefreshProfiles, NoticeMessage { status: &'a str, message: String }, ProfileChanged { current_profile_id: &'a String }, TimerUpdated { profile_index: &'a String }, @@ -34,6 +35,7 @@ impl NotificationSystem { match event { FrontendEvent::RefreshClash => ("verge://refresh-clash-config", Ok(json!("yes"))), FrontendEvent::RefreshVerge => ("verge://refresh-verge-config", Ok(json!("yes"))), + FrontendEvent::RefreshProfiles => ("verge://refresh-profiles", Ok(json!("yes"))), FrontendEvent::NoticeMessage { status, message } => { ("verge://notice-message", serde_json::to_value((status, message))) } diff --git a/src/providers/app-data-provider.tsx b/src/providers/app-data-provider.tsx index 4ccb883f1..06fc86ee3 100644 --- a/src/providers/app-data-provider.tsx +++ b/src/providers/app-data-provider.tsx @@ -155,6 +155,10 @@ export const AppDataProvider = ({ refreshProxy().catch(() => {}) } + const handleRefreshProfiles = () => { + void revalidateQueries([['getProfiles']]) + } + const initializeListeners = async () => { try { const unlistenProfile = await listen( @@ -166,6 +170,16 @@ export const AppDataProvider = ({ console.error('[AppDataProvider] 监听 Profile 事件失败:', error) } + try { + const unlistenProfiles = await listen( + 'verge://refresh-profiles', + handleRefreshProfiles, + ) + cleanupFns.push(unlistenProfiles) + } catch (error) { + console.error('[AppDataProvider] 监听 Profiles 刷新事件失败:', error) + } + try { const unlistenProxy = await listen( 'verge://refresh-proxy-config',