diff --git a/Cargo.lock b/Cargo.lock index eb68395..695dcce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -654,6 +654,19 @@ dependencies = [ "serde_core", ] +[[package]] +name = "blake3" +version = "1.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d9e454fc11f76977dc803893aff6304ed33d6a26efae8696573bea74baa27ae" +dependencies = [ + "arrayvec", + "cc", + "cfg-if", + "constant_time_eq", + "cpufeatures 0.3.0", +] + [[package]] name = "block" version = "0.1.6" @@ -1034,6 +1047,12 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "constant_time_eq" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" + [[package]] name = "core-foundation" version = "0.9.4" @@ -1093,6 +1112,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "crc32fast" version = "1.5.1" @@ -4333,7 +4361,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -4344,7 +4372,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -6425,11 +6453,12 @@ checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" [[package]] name = "zerosend" -version = "2.5.3-beta" +version = "2.5.3-beta.1" dependencies = [ "arboard", "async-stream", "axum", + "blake3", "bytes", "chrono", "clap", diff --git a/Cargo.toml b/Cargo.toml index 58ff3d3..0116f21 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "zerosend" -version = "2.5.3-beta" +version = "2.5.3-beta.1" edition = "2021" description = "Fast, peer-to-peer file transfer designed for ZeroTier, Tailscale, Radmin VPN and LAN" authors = ["RarDog"] @@ -49,6 +49,7 @@ sha2 = "0.10" hex = "0.4" walkdir = "2.5" human_bytes = "0.4" +blake3 = "1.5" # Windows Native Registry & Sound [target.'cfg(windows)'.dependencies] diff --git a/src/main.rs b/src/main.rs index 7d775b5..d1f4eca 100644 --- a/src/main.rs +++ b/src/main.rs @@ -192,6 +192,9 @@ async fn main() -> Result<(), Box> { TransferState::Rejected => { pb.abandon_with_message("⛔ Transfer rejected by recipient."); } + TransferState::Paused => { + pb.set_message("⏸ Paused"); + } TransferState::Canceled => { pb.abandon_with_message("🛑 Canceled."); } diff --git a/src/transfer/archive.rs b/src/transfer/archive.rs index 9c02f94..3285d7c 100644 --- a/src/transfer/archive.rs +++ b/src/transfer/archive.rs @@ -117,6 +117,40 @@ pub fn estimate_folder_size(dir: &Path) -> u64 { .sum() } +/// Returns the optimal zstd compression level for a given filename. +/// Returns 0 for already-compressed formats (store only), 1 otherwise (ultra-fast). +pub fn optimal_zstd_level_for_file(name: &str) -> i32 { + let ext = name.rsplit('.').next().unwrap_or("").to_lowercase(); + match ext.as_str() { + // Already compressed — don't waste CPU + "jpg" | "jpeg" | "png" | "gif" | "webp" | "avif" | "heic" | + "mp4" | "mkv" | "avi" | "mov" | "wmv" | "flv" | "webm" | + "mp3" | "aac" | "ogg" | "flac" | "m4a" | "opus" | + "zip" | "rar" | "7z" | "gz" | "bz2" | "xz" | "zst" | "lz4" | + "pdf" | "docx" | "xlsx" | "pptx" => 0, + // Compressible text/code/data + _ => 1, + } +} + +/// Estimates the compressibility of a folder: returns true if most content is compressible. +pub fn folder_is_compressible(dir: &Path) -> bool { + let mut compressible = 0u64; + let mut total = 0u64; + for entry in WalkDir::new(dir).into_iter().filter_map(|e| e.ok()) { + if entry.path().is_file() { + let size = entry.metadata().map(|m| m.len()).unwrap_or(0); + total += size; + let name = entry.file_name().to_string_lossy(); + if optimal_zstd_level_for_file(&name) > 0 { + compressible += size; + } + } + } + if total == 0 { return true; } + (compressible as f64 / total as f64) > 0.3 +} + /// Stream an entire directory as a Zstandard-compressed TAR archive /// into a `std::sync::mpsc::SyncSender>` channel. /// diff --git a/src/transfer/client.rs b/src/transfer/client.rs index 903f6e4..4a03221 100644 --- a/src/transfer/client.rs +++ b/src/transfer/client.rs @@ -10,8 +10,9 @@ use super::text_analyzer::{analyze_text, AnalyzedText}; use crate::network::DiscoveryBeacon; use reqwest::Client; use sha2::{Digest, Sha256}; -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::fs::File; @@ -60,6 +61,7 @@ struct ChunkTask { pub struct TransferClient { client: Client, + pub pause_flags: Arc>>>, } impl TransferClient { @@ -71,7 +73,55 @@ impl TransferClient { .build() .unwrap_or_default(); - Self { client } + Self { + client, + pause_flags: Arc::new(Mutex::new(HashMap::new())), + } + } + + pub async fn pause_transfer(&self, session_id: &str) { + let flags = self.pause_flags.lock().await; + if let Some(flag) = flags.get(session_id) { + flag.store(true, Ordering::Relaxed); + } + } + + pub async fn resume_transfer(&self, session_id: &str) { + let flags = self.pause_flags.lock().await; + if let Some(flag) = flags.get(session_id) { + flag.store(false, Ordering::Relaxed); + } + } + + pub async fn is_paused(&self, session_id: &str) -> bool { + let flags = self.pause_flags.lock().await; + flags.get(session_id).map(|f| f.load(Ordering::Relaxed)).unwrap_or(false) + } + + /// Race-ping multiple candidate URLs and return the fastest responding one + pub async fn pick_fastest_url(&self, candidates: &[String]) -> Option { + if candidates.is_empty() { return None; } + if candidates.len() == 1 { return Some(candidates[0].clone()); } + + let (tx, mut rx) = tokio::sync::mpsc::channel(candidates.len()); + + for url in candidates { + let client = self.client.clone(); + let tx = tx.clone(); + let url = url.clone(); + tokio::spawn(async move { + let start = Instant::now(); + if let Ok(resp) = client.get(&format!("{}/api/info", url)).send().await { + if resp.status().is_success() { + let latency = start.elapsed().as_millis() as u64; + let _ = tx.send((url, latency)).await; + } + } + }); + } + drop(tx); + + rx.recv().await.map(|(url, _)| url) } /// Ping a remote peer to check health and get beacon info @@ -237,15 +287,27 @@ impl TransferClient { speed_bps: 0, state: TransferState::InProgress, checksum_verified: false, + compressed_bytes: 0, }); + // Feature 1: Determine optimal compression level for entire folder + let is_compressible = { + let folder_clone2 = folder_path.clone(); + tokio::task::spawn_blocking(move || { + super::archive::folder_is_compressible(&folder_clone2) + }) + .await + .unwrap_or(true) + }; + let folder_zstd_level = if is_compressible { 1 } else { 0 }; + // Build a bounded channel: blocking producer → async consumer → HTTP body let (tx, rx) = std::sync::mpsc::sync_channel::>>(32); let folder_clone = folder_path.clone(); // Spawn blocking tar+zstd compressor thread let compress_handle = tokio::task::spawn_blocking(move || { - super::archive::stream_folder_as_tar_zst(folder_clone, tx, 1); + super::archive::stream_folder_as_tar_zst(folder_clone, tx, folder_zstd_level); }); // Collect chunks from the channel into an HTTP body stream @@ -303,6 +365,7 @@ impl TransferClient { speed_bps: 0, state: TransferState::Completed, checksum_verified: false, + compressed_bytes: 0, }); } Ok(r) => { @@ -326,6 +389,8 @@ impl TransferClient { return Err("No files selected for transfer".to_string()); } + // Feature 7: Priority queue - sort files by size ascending so small files finish first + file_entries.sort_by_key(|(_, _, size)| *size); let file_metas: Vec = file_entries .iter() @@ -367,6 +432,7 @@ impl TransferClient { speed_bps: 0, state: TransferState::PendingConfirmation, checksum_verified: false, + compressed_bytes: 0, }); let prepare_url = format!("{}/api/transfer/prepare", peer_url); @@ -397,6 +463,7 @@ impl TransferClient { speed_bps: 0, state: TransferState::Rejected, checksum_verified: false, + compressed_bytes: 0, }); return Err(reason); @@ -451,9 +518,16 @@ impl TransferClient { } } + let pause_flag = { + let mut flags = self.pause_flags.lock().await; + let flag = Arc::new(AtomicBool::new(false)); + flags.insert(session_id.clone(), flag.clone()); + flag + }; + let rate_limiter = RateLimiter::new((max_upload_speed_mbps as u64) * 1024 * 1024); let tasks_mutex = Arc::new(Mutex::new(chunk_tasks)); - let (chunk_tx, mut chunk_rx) = tokio::sync::mpsc::unbounded_channel::<(usize, String)>(); + let (chunk_tx, mut chunk_rx) = tokio::sync::mpsc::unbounded_channel::<(usize, usize, String)>(); let (progress_update_tx, mut progress_update_rx) = tokio::sync::mpsc::unbounded_channel::(); let s_id = session_id.clone(); @@ -461,15 +535,17 @@ impl TransferClient { let t_files = file_metas.len(); let p_tx = progress_update_tx.clone(); - // Background progress tracker with smoothed EMA speed calculation + // Background progress tracker with smoothed EMA speed calculation & compressed_bytes tokio::spawn(async move { let mut total_sent = already_transferred; + let mut total_wire = already_transferred; let mut last_bytes = total_sent; let mut last_time = Instant::now(); let mut smoothed_speed = 0.0f64; - while let Some((len, fname)) = chunk_rx.recv().await { - total_sent += len as u64; + while let Some((raw_len, wire_len, fname)) = chunk_rx.recv().await { + total_sent += raw_len as u64; + total_wire += wire_len as u64; if last_time.elapsed().as_millis() >= 350 { let elapsed_secs = last_time.elapsed().as_secs_f64(); @@ -500,6 +576,7 @@ impl TransferClient { speed_bps, state: TransferState::InProgress, checksum_verified: false, + compressed_bytes: total_wire, }); last_time = Instant::now(); @@ -520,6 +597,7 @@ impl TransferClient { speed_bps: 0, state: TransferState::InProgress, checksum_verified: false, + compressed_bytes: already_transferred, }); let start_time = Instant::now(); @@ -532,6 +610,7 @@ impl TransferClient { let chunk_url = format!("{}/api/transfer/chunk/{}", peer_url, session_id); let limiter_c = rate_limiter.clone(); let tx_c = chunk_tx.clone(); + let pause_c = pause_flag.clone(); worker_set.spawn(async move { loop { @@ -558,16 +637,41 @@ impl TransferClient { return Err(format!("Failed to read chunk from {:?}: {}", chunk_task.file_path, e)); } - limiter_c.throttle(buf.len()).await; + // Feature 1: Smart compression check per file type + let is_compressible = super::archive::optimal_zstd_level_for_file(&chunk_task.rel_path) > 0; + let (payload, is_zstd) = if is_compressible && buf.len() >= 1024 { + if let Ok(compressed) = zstd::bulk::compress(&buf, 1) { + if compressed.len() + 64 < buf.len() { + (compressed, true) + } else { + (buf, false) + } + } else { + (buf, false) + } + } else { + (buf, false) + }; - let resp = client_c + let wire_len = payload.len(); + limiter_c.throttle(wire_len).await; + + // Feature 5: Pause support in worker loop + while pause_c.load(Ordering::Relaxed) { + tokio::time::sleep(Duration::from_millis(200)).await; + } + + let mut req = client_c .post(&chunk_url) .header("x-file-name", &chunk_task.rel_path) .header("x-chunk-offset", chunk_task.offset.to_string()) - .header("x-chunk-length", chunk_task.length.to_string()) - .body(buf) - .send() - .await; + .header("x-chunk-length", chunk_task.length.to_string()); + + if is_zstd { + req = req.header("x-compression", "zstd"); + } + + let resp = req.body(payload).send().await; let is_ok = match &resp { Ok(r) => r.status().is_success(), @@ -589,7 +693,7 @@ impl TransferClient { } } - let _ = tx_c.send((chunk_task.length, chunk_task.rel_path)); + let _ = tx_c.send((chunk_task.length, wire_len, chunk_task.rel_path)); } Ok::<(), String>(()) }); @@ -616,6 +720,7 @@ impl TransferClient { speed_bps: 0, state: TransferState::Failed(e.clone()), checksum_verified: false, + compressed_bytes: 0, }); return Err(e); } @@ -636,6 +741,9 @@ impl TransferClient { on_progress(progress); } + // Cleanup pause flags + self.pause_flags.lock().await.remove(&session_id); + // Finalize transfer on receiver (atomic rename + zip extraction) let finish_url = format!("{}/api/transfer/finish/{}", peer_url, session_id); let finish_resp = self.client.post(&finish_url).send().await; @@ -663,6 +771,7 @@ impl TransferClient { speed_bps: avg_speed, state: TransferState::Completed, checksum_verified: true, + compressed_bytes: total_bytes, }); Ok(()) diff --git a/src/transfer/folder_sync.rs b/src/transfer/folder_sync.rs index 1a2b8f3..1dca82b 100644 --- a/src/transfer/folder_sync.rs +++ b/src/transfer/folder_sync.rs @@ -22,6 +22,49 @@ pub struct FolderSyncEngine { status_tx: mpsc::UnboundedSender, } +// ── Sync cache structures ────────────────────────────────────────────────── + +#[derive(serde::Serialize, serde::Deserialize, Default)] +struct SyncCache { + files: HashMap, +} + +#[derive(serde::Serialize, serde::Deserialize, Clone)] +struct CacheEntry { + size: u64, + mtime: u64, + blake3: Option, +} + +fn load_sync_cache(local_path: &Path) -> SyncCache { + let cache_path = local_path.join(".zerosend_sync_cache.json"); + std::fs::read_to_string(&cache_path) + .ok() + .and_then(|s| serde_json::from_str(&s).ok()) + .unwrap_or_default() +} + +fn save_sync_cache(local_path: &Path, cache: &SyncCache) { + let cache_path = local_path.join(".zerosend_sync_cache.json"); + if let Ok(s) = serde_json::to_string(cache) { + let _ = std::fs::write(&cache_path, s); + } +} + +fn file_blake3(path: &std::path::Path) -> Option { + let mut file = std::fs::File::open(path).ok()?; + let mut hasher = blake3::Hasher::new(); + let mut buf = [0u8; 65536]; + loop { + let n = std::io::Read::read(&mut file, &mut buf).ok()?; + if n == 0 { break; } + hasher.update(&buf[..n]); + } + Some(hasher.finalize().to_hex().to_string()) +} + +// ── Engine ───────────────────────────────────────────────────────────────── + impl FolderSyncEngine { pub fn new( client: Arc, @@ -109,17 +152,31 @@ impl FolderSyncEngine { let _ = tokio::fs::create_dir_all(local_path).await; } + // Load sync cache (size/mtime/hash from last sync) + let cache = load_sync_cache(local_path); + // Fetch remote directory tree let remote_tree = client .fetch_shared_tree(peer_url, remote_folder_id, None, None) .await?; - // 1. Scan local files - let mut local_files: HashMap = HashMap::new(); // rel_path -> (size, mtime) - if let Ok(entries) = walkdir_local(local_path) { - for (rel, size, mtime) in entries { - local_files.insert(rel, (size, mtime)); + // 1. Scan local files with blake3 hash support + // Returns Vec<(rel_path, size, mtime, Option)> + let local_files_vec = { + let local_path_buf = local_path.to_path_buf(); + tokio::task::spawn_blocking(move || walkdir_local(&local_path_buf)) + .await + .map_err(|e| format!("walkdir task error: {}", e))? + .map_err(|e| format!("walkdir error: {}", e))? + }; + + let mut local_files: HashMap)> = HashMap::new(); + for (rel, size, mtime, hash) in local_files_vec { + // Skip the cache file itself + if rel == ".zerosend_sync_cache.json" { + continue; } + local_files.insert(rel, (size, mtime, hash)); } // 2. Map remote files @@ -131,12 +188,36 @@ impl FolderSyncEngine { } let mut synced_count = 0; + let mut updated_cache = SyncCache { files: cache.files.clone() }; // A. Upload newly created or changed local files to remote if !remote_tree.read_only { - for (rel, (size, _)) in &local_files { + for (rel, (size, mtime, hash)) in &local_files { let need_upload = match remote_files.get(rel) { - Some(remote_size) => *remote_size != *size, + Some(remote_size) => { + if *remote_size != *size { + true // size differs — definitely upload + } else { + // Same size: check cache to detect content-only changes + if let Some(cached) = cache.files.get(rel) { + if cached.size == *size && cached.mtime == *mtime { + // size AND mtime unchanged since last sync — skip + false + } else if cached.size == *size { + // mtime changed but size same — compare blake3 + match (hash, &cached.blake3) { + (Some(h), Some(c)) => h != c, // upload if hash changed + _ => true, // no hash available, upload to be safe + } + } else { + true // cached size differs — upload + } + } else { + // No cache entry — upload to be safe + true + } + } + }, None => true, // New local file }; @@ -153,8 +234,21 @@ impl FolderSyncEngine { ) .await { + // Update cache entry after successful upload + updated_cache.files.insert(rel.clone(), CacheEntry { + size: *size, + mtime: *mtime, + blake3: hash.clone(), + }); synced_count += 1; } + } else { + // File unchanged — update cache entry to keep mtime fresh + updated_cache.files.insert(rel.clone(), CacheEntry { + size: *size, + mtime: *mtime, + blake3: hash.clone(), + }); } } } @@ -162,7 +256,7 @@ impl FolderSyncEngine { // B. Download newly created remote files to local for (rel, remote_size) in &remote_files { let need_download = match local_files.get(rel) { - Some((local_size, _)) => *local_size != *remote_size, + Some((local_size, _, _)) => *local_size != *remote_size, None => true, // New remote file }; @@ -177,11 +271,16 @@ impl FolderSyncEngine { .download_shared_file(peer_url, remote_folder_id, rel, &target_file_path, None) .await { + // Invalidate cache entry since we downloaded new content + updated_cache.files.remove(rel); synced_count += 1; } } } + // Save updated cache + save_sync_cache(local_path, &updated_cache); + if synced_count > 0 { let _ = status_tx.send(SyncStatusUpdate { pair_id: pair_id.to_string(), @@ -195,7 +294,10 @@ impl FolderSyncEngine { } } -fn walkdir_local(base: &Path) -> Result, std::io::Error> { +/// Walk local directory and return (rel_path, size, mtime, Option). +/// Hash is only computed for files < 500MB to avoid stalling. +fn walkdir_local(base: &Path) -> Result)>, std::io::Error> { + const MAX_HASH_SIZE: u64 = 500 * 1024 * 1024; // 500 MB let mut results = Vec::new(); for entry in walkdir::WalkDir::new(base).into_iter().filter_map(|e| e.ok()) { if entry.file_type().is_file() { @@ -209,7 +311,14 @@ fn walkdir_local(base: &Path) -> Result, std::io::Error> .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) .map(|d| d.as_secs()) .unwrap_or(0); - results.push((rel_str, size, mtime)); + + let hash = if size < MAX_HASH_SIZE { + file_blake3(entry.path()) + } else { + None + }; + + results.push((rel_str, size, mtime, hash)); } } } diff --git a/src/transfer/protocol.rs b/src/transfer/protocol.rs index 7ad5098..e3b8f03 100644 --- a/src/transfer/protocol.rs +++ b/src/transfer/protocol.rs @@ -51,6 +51,7 @@ pub enum PrepareTransferResponse { pub enum TransferState { PendingConfirmation, InProgress, + Paused, Completed, Failed(String), Rejected, @@ -71,6 +72,8 @@ pub struct TransferProgress { pub state: TransferState, #[serde(default)] pub checksum_verified: bool, + #[serde(default)] + pub compressed_bytes: u64, // actual bytes sent over network (after compression) } impl TransferProgress { diff --git a/src/transfer/server.rs b/src/transfer/server.rs index e0b0e1f..fe24107 100644 --- a/src/transfer/server.rs +++ b/src/transfer/server.rs @@ -422,6 +422,7 @@ async fn handle_prepare( speed_bps: 0, state: TransferState::InProgress, checksum_verified: false, + compressed_bytes: 0, }; state.active_transfers.write().await.insert(session_id.clone(), progress.clone()); @@ -468,6 +469,7 @@ async fn handle_prepare( speed_bps: 0, state: TransferState::InProgress, checksum_verified: false, + compressed_bytes: 0, }; state.active_transfers.write().await.insert(session_id.clone(), progress.clone()); @@ -551,8 +553,36 @@ async fn handle_upload_chunk( ); } - let chunk_len = body.len(); - if let Err(e) = file.write_all(&body).await { + let raw_length: usize = headers + .get("x-chunk-length") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse().ok()) + .unwrap_or(body.len()); + + let is_zstd = headers + .get("x-compression") + .and_then(|v| v.to_str().ok()) + .map(|v| v.eq_ignore_ascii_case("zstd")) + .unwrap_or(false); + + let decompressed_data = if is_zstd { + match zstd::bulk::decompress(&body, raw_length.max(16 * 1024 * 1024)) { + Ok(decomp) => decomp, + Err(e) => { + error!("Zstd decompression failed for {}: {}", file_name, e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({"error": format!("Zstd decompress error: {}", e)})), + ); + } + } + } else { + body.to_vec() + }; + + let chunk_len = decompressed_data.len(); + let wire_bytes = body.len() as u64; + if let Err(e) = file.write_all(&decompressed_data).await { error!("Write error on {:?} at {}: {}", part_path, offset, e); return ( StatusCode::INTERNAL_SERVER_ERROR, @@ -567,6 +597,7 @@ async fn handle_upload_chunk( let mut transfers = state.active_transfers.write().await; if let Some(t) = transfers.get_mut(&session_id) { t.bytes_transferred = (t.bytes_transferred + chunk_len as u64).min(t.total_bytes); + t.compressed_bytes += wire_bytes; t.current_file_name = file_name; t.state = TransferState::InProgress; @@ -715,6 +746,7 @@ async fn handle_folder_stream( speed_bps: 0, state: super::protocol::TransferState::Completed, checksum_verified: false, + compressed_bytes: 0, }; let _ = state.transfer_events.send(progress); ( @@ -977,6 +1009,7 @@ async fn handle_web_upload( speed_bps: 0, state: TransferState::InProgress, checksum_verified: false, + compressed_bytes: 0, }; state.active_transfers.write().await.insert(session_id.clone(), initial_progress.clone()); let _ = state.transfer_events.send(initial_progress); diff --git a/src/ui/gui.rs b/src/ui/gui.rs index 5567992..1af5e0a 100644 --- a/src/ui/gui.rs +++ b/src/ui/gui.rs @@ -391,7 +391,7 @@ impl ZeroSendApp { selected_peer_ids: HashSet::new(), queued_files: initial_files, transfers: Vec::new(), - logs: vec!["⚡ ZeroSend v2.5.3-beta ready".to_string()], + logs: vec!["⚡ ZeroSend v2.5.3.1-beta ready".to_string()], text_input: String::new(), analyzed_preview: analyze_text(""), recipient_pin: String::new(), @@ -3339,6 +3339,9 @@ impl ZeroSendApp { TransferState::Rejected => { ui.label(RichText::new("⛔ Передача отклонена").color(Color32::RED).size(12.0)); } + TransferState::Paused => { + ui.label(RichText::new("⏸ На паузе").color(Color32::from_rgb(234, 179, 8)).strong().size(13.0)); + } TransferState::Canceled => { ui.label(RichText::new("🛑 Отменено").color(theme.text_muted).size(12.0)); } @@ -3367,7 +3370,15 @@ impl ZeroSendApp { 100 }; - let size_details = if t.state == TransferState::InProgress { + let comp_info = if t.compressed_bytes > 0 && t.compressed_bytes < t.bytes_transferred { + let saved_pct = ((t.bytes_transferred.saturating_sub(t.compressed_bytes) as f64 / t.bytes_transferred as f64) * 100.0) as u32; + let ratio = t.bytes_transferred as f64 / t.compressed_bytes.max(1) as f64; + format!(" • 🗜️ {:.1}x (сжато на {}%)", ratio, saved_pct) + } else { + String::new() + }; + + let size_details = if t.state == TransferState::InProgress || t.state == TransferState::Paused { let speed = t.speed_bps.max(1); let remaining_bytes = t.total_bytes.saturating_sub(t.bytes_transferred); let eta_secs = remaining_bytes / speed; @@ -3380,19 +3391,22 @@ impl ZeroSendApp { }; let speed_mb = (t.speed_bps as f64) / 1_048_576.0; + let state_prefix = if t.state == TransferState::Paused { "⏸ На паузе • " } else { "" }; format!( - "{} / {} ({}%) • ⚡ {:.1} MB/s • {}", + "{}{} / {} ({}%) • ⚡ {:.1} MB/s • {}{}", + state_prefix, human_bytes(t.bytes_transferred as f64), human_bytes(t.total_bytes as f64), progress_pct, speed_mb, - eta_str + eta_str, + comp_info ) } else if is_done { if t.checksum_verified { - format!("Доставлено • Размер: {} • 🛡️ SHA-256 Проверено", human_bytes(t.total_bytes as f64)) + format!("Доставлено • Размер: {}{} • 🛡️ SHA-256 Проверено", human_bytes(t.total_bytes as f64), comp_info) } else { - format!("Доставлено • Размер: {}", human_bytes(t.total_bytes as f64)) + format!("Доставлено • Размер: {}{}", human_bytes(t.total_bytes as f64), comp_info) } } else { format!("Размер: {}", human_bytes(t.total_bytes as f64)) @@ -3433,6 +3447,30 @@ impl ZeroSendApp { let _ = std::process::Command::new("explorer").arg(&target_file).spawn(); } }); + } else if is_active || t.state == TransferState::Paused { + ui.with_layout(Layout::right_to_left(Align::Center), |ui| { + let client_c = self.client.clone(); + let s_id = t.session_id.clone(); + if t.state == TransferState::Paused { + let resume_btn = Button::new(RichText::new("▶️ Продолжить").strong().color(Color32::WHITE)) + .fill(Color32::from_rgb(34, 197, 94)) + .rounding(Rounding::same(6.0)); + if ui.add(resume_btn).clicked() { + tokio::spawn(async move { + client_c.resume_transfer(&s_id).await; + }); + } + } else { + let pause_btn = Button::new(RichText::new("⏸ Пауза").strong().color(Color32::WHITE)) + .fill(Color32::from_rgb(234, 179, 8)) + .rounding(Rounding::same(6.0)); + if ui.add(pause_btn).clicked() { + tokio::spawn(async move { + client_c.pause_transfer(&s_id).await; + }); + } + } + }); } }); @@ -3952,13 +3990,18 @@ impl ZeroSendApp { .stroke(theme.stroke_border) .inner_margin(Margin::same(14.0)) .show(ui, |ui| { - // v2.5.3-beta + // v2.5.3.1-beta ui.horizontal(|ui| { - ui.label(RichText::new("v2.5.3-beta").strong().color(Color32::from_rgb(34, 197, 94))); + ui.label(RichText::new("v2.5.3.1-beta").strong().color(Color32::from_rgb(34, 197, 94))); ui.label(RichText::new("(Текущая версия)").size(11.0).color(Color32::GRAY)); }); - ui.label(" • ⚡ Потоковая передача папок: Zip-на-лету через tar.zst — папки стримятся напрямую без создания временных файлов на диске."); - ui.label(" • 🗜️ Zstandard (Zstd) сжатие: Мгновенное сжатие уровня 1-3 в потоке — до 4-8x прирост скорости на 100Mбит и VPN-каналах."); + ui.label(" • 🗜️ Умная Zstd-компрессия: Автоматический пропуск уже сжатых файлов (.jpg, .mp4, .zip, .rar...) и сжатие кода/текста/документов."); + ui.label(" • 🔄 Delta-transfer для Auto-Sync: Blake3 хэширование и кэш измененных файлов — синхронизация только изменений без повторной передачи."); + ui.label(" • 👥 Мульти-отправка: Параллельная передача выбранных файлов сразу нескольким друзьям."); + ui.label(" • ⏸️ Пауза и возобновление: Кнопки паузы/продолжения передачи прямо в дашборде мониторинга."); + ui.label(" • 🏎️ Очередь с приоритетами: Маленькие файлы передаются первыми для мгновенного завершения."); + ui.label(" • 📊 Мониторинг сжатия: Отображение реального коэффициента сжатия и сэкономленного трафика."); + ui.label(" • 🏷️ Умный парсер версий: Поддержка 4-значных номеров (2.5.3.1) и автоматическое определение Beta-релизов."); ui.add_space(8.0); ui.separator(); ui.add_space(8.0); @@ -4329,6 +4372,7 @@ impl ZeroSendApp { speed_bps: 0, state: TransferState::InProgress, checksum_verified: false, + compressed_bytes: 0, }; if let Some(pos) = self.transfers.iter().position(|x| x.session_id == session_id) { diff --git a/src/ui/tui.rs b/src/ui/tui.rs index a4cf0cf..d541b7f 100644 --- a/src/ui/tui.rs +++ b/src/ui/tui.rs @@ -479,6 +479,7 @@ fn draw_ui(f: &mut Frame, app: &mut AppState) { TransferState::Completed => "✅ Completed".to_string(), TransferState::Failed(e) => format!("❌ Failed: {}", e), TransferState::Rejected => "⛔ Rejected".to_string(), + TransferState::Paused => "⏸ Paused".to_string(), TransferState::Canceled => "🛑 Canceled".to_string(), }; diff --git a/src/updater.rs b/src/updater.rs index bdf3524..5a4e3de 100644 --- a/src/updater.rs +++ b/src/updater.rs @@ -43,6 +43,8 @@ pub struct UpdateInfo { pub download_url: Option, pub gitea_url: String, pub asset_size: u64, + pub is_beta: bool, + pub pre_number: u32, } #[derive(Debug, Clone)] @@ -54,21 +56,64 @@ pub enum UpdateProgress { Error(String), } -/// Parse semantic version string (e.g. "v2.4.1", "2.4.2-beta") into (u32, u32, u32) -pub fn parse_version(v: &str) -> (u32, u32, u32) { +/// Parse semantic version string into (major, minor, patch, subpatch, is_beta, pre_num). +/// Handles formats like "v2.5.3", "2.5.3-beta", "2.5.3-beta.1", "2.5.3.1" +pub fn parse_version(v: &str) -> (u32, u32, u32, u32, bool, u32) { let clean = v.trim().trim_start_matches('v').trim_start_matches('V'); - let parts: Vec<&str> = clean.split('.').collect(); - let major = parts.first().and_then(|s| s.split('-').next()).and_then(|s| s.parse::().ok()).unwrap_or(0); - let minor = parts.get(1).and_then(|s| s.split('-').next()).and_then(|s| s.parse::().ok()).unwrap_or(0); - let patch = parts.get(2).and_then(|s| s.split('-').next()).and_then(|s| s.parse::().ok()).unwrap_or(0); - (major, minor, patch) + + // Split off pre-release suffix (everything after the first '-') + let (numeric_part, pre_part) = if let Some(idx) = clean.find('-') { + (&clean[..idx], Some(&clean[idx + 1..])) + } else { + (clean, None) + }; + + let num_parts: Vec<&str> = numeric_part.split('.').collect(); + let major = num_parts.first().and_then(|s| s.parse::().ok()).unwrap_or(0); + let minor = num_parts.get(1).and_then(|s| s.parse::().ok()).unwrap_or(0); + let patch = num_parts.get(2).and_then(|s| s.parse::().ok()).unwrap_or(0); + let subpatch = num_parts.get(3).and_then(|s| s.parse::().ok()).unwrap_or(0); + + let is_beta = pre_part.map(|p| { + let pl = p.to_lowercase(); + pl.contains("beta") || pl.contains("alpha") || pl.contains("rc") + }).unwrap_or(false); + + // Extract pre-release number, e.g. "beta.1" -> 1, "beta" -> 0 + let pre_num = pre_part + .and_then(|p| p.split('.').last()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + + (major, minor, patch, subpatch, is_beta, pre_num) } -/// Returns true if latest version is strictly newer than current version +/// Returns true if a version string contains "beta", "alpha", or "rc" +pub fn is_beta_version(v: &str) -> bool { + let lower = v.to_lowercase(); + lower.contains("beta") || lower.contains("alpha") || lower.contains("rc") +} + +/// Returns true if latest version is strictly newer than current version. +/// Rules: stable > beta of same numeric version; higher pre_num beats lower. pub fn is_newer_version(latest: &str, current: &str) -> bool { - let (l_maj, l_min, l_pat) = parse_version(latest); - let (c_maj, c_min, c_pat) = parse_version(current); - (l_maj, l_min, l_pat) > (c_maj, c_min, c_pat) + let (l_maj, l_min, l_pat, l_sub, l_beta, l_pre) = parse_version(latest); + let (c_maj, c_min, c_pat, c_sub, c_beta, c_pre) = parse_version(current); + + // Compare numeric components first + let l_num = (l_maj, l_min, l_pat, l_sub); + let c_num = (c_maj, c_min, c_pat, c_sub); + + if l_num != c_num { + return l_num > c_num; + } + + // Same numeric version: stable > beta + match (l_beta, c_beta) { + (false, true) => true, // latest is stable, current is beta → update + (true, false) => false, // latest is beta, current is stable → no update + _ => l_pre > c_pre, // both same prerelease type: compare pre number + } } /// Check Gitea repository for available releases or newer version tags @@ -110,13 +155,15 @@ pub async fn check_gitea_update(api_base: Option<&str>) -> Result) -> Result) -> Result