19 Commits
Author SHA1 Message Date
RarDog 1e2b4706d5 Release v2.5.6: window position and size persistence on launch and close 2026-08-28 14:47:27 +03:00
RarDog c3f26b77e8 Updated modern Send tab and Network Shares explorer in v2.5.5 2026-08-28 14:42:53 +03:00
RarDog 9957c9b70f Revert UI design changes back to original clean layout in v2.5.5 2026-08-28 14:35:07 +03:00
RarDog d4f52e3ec6 Completely eliminate panel separation gap with unified horizontal container in v2.5.5 2026-08-28 14:29:55 +03:00
RarDog 3ca52e2649 Harmonize theme backgrounds to eliminate dark gap between panels in v2.5.5 2026-08-28 14:24:52 +03:00
RarDog 90fda425bd Fix black separator line between sidebar and main content in v2.5.5 2026-08-28 14:20:27 +03:00
RarDog 124d9c2dd4 Update Send tab GUI, rich ZeroTier cards, and copyable IP in v2.5.5 2026-08-28 14:15:48 +03:00
RarDog e019ddccc9 Release v2.5.5: beautiful network shares UI, breadcrumbs navigation & instant transfers tab redirection on shared downloads 2026-08-28 14:06:53 +03:00
RarDog 61f2545588 Release v2.5.4: live shared folder discovery, shared download tracking, streaming folder progress & persistent transfer history 2026-08-28 13:56:38 +03:00
RarDog b751d82748 Release v2.5.3-beta.2: fluid tab transition animations with cubic ease-out and animated sidebar indicator 2026-08-27 17:35:15 +03:00
RarDog c27ac0f41c Audit improvements: fix Windows atomic rename collisions and clean up compiler warnings 2026-08-27 15:55:06 +03:00
RarDog f65b15452f Release v2.5.3-beta.1.1: 4MB TCP socket buffer tuning, Zero-Drop auto-recovery, and mDNS multicast discovery 2026-08-27 15:41:12 +03:00
RarDog 686eb2fb6f Release v2.5.3.1-beta: smart Zstd compression, Blake3 delta-sync, multi-send, pause/resume, priority queue, and 4-part version parser 2026-08-27 15:22:02 +03:00
RarDog 0867b85018 Fix: update settings checkbox and send panel labels from ZIP to tar.zst streaming 2026-08-27 13:44:00 +03:00
RarDog a6a32bb23f Release v2.5.3-beta: streaming tar.zst folder transfer + Zstandard compression 2026-08-27 13:38:32 +03:00
RarDog 35e590c718 Release v2.5.2: fix transfers speed metric overflow, auto-cleanup old binary on update, full Russian localization 2026-08-27 13:18:55 +03:00
RarDog 9f9044646f UI: Complete modernization of all GUI tabs (Send, Receive, Transfers, Settings) 2026-08-27 13:09:37 +03:00
RarDog dc72fe47e1 feat: ZeroSend v2.5.1 - Revamped 3-Step Live Folder Auto-Sync, Dynamic Credits & UI Polish 2026-08-27 12:59:46 +03:00
RarDog 514f88ae33 feat: ZeroSend v2.5.0 - Live Folder Sync, WebDAV Drive (Z:), Watch Party, Full Vim Editor & Remote Grep 2026-08-27 12:46:28 +03:00
16 changed files with 5113 additions and 1013 deletions
Generated
+119 -3
View File
@@ -440,6 +440,28 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "async-stream"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
dependencies = [
"async-stream-impl",
"futures-core",
"pin-project-lite",
]
[[package]]
name = "async-stream-impl"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "async-task"
version = "4.7.1"
@@ -632,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"
@@ -1012,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"
@@ -1071,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"
@@ -1544,6 +1594,16 @@ dependencies = [
"rustc_version",
]
[[package]]
name = "filetime"
version = "0.2.29"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759"
dependencies = [
"cfg-if",
"libc",
]
[[package]]
name = "find-msvc-tools"
version = "0.1.11"
@@ -4301,7 +4361,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8"
dependencies = [
"cfg-if",
"cpufeatures",
"cpufeatures 0.2.17",
"digest",
]
@@ -4312,7 +4372,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
dependencies = [
"cfg-if",
"cpufeatures",
"cpufeatures 0.2.17",
"digest",
]
@@ -4651,6 +4711,17 @@ dependencies = [
"version-compare",
]
[[package]]
name = "tar"
version = "0.4.46"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840"
dependencies = [
"filetime",
"libc",
"xattr",
]
[[package]]
name = "target-lexicon"
version = "0.12.16"
@@ -6090,6 +6161,16 @@ version = "0.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ea6fc2961e4ef194dcbfe56bb845534d0dc8098940c7e5c012a258bfec6701bd"
[[package]]
name = "xattr"
version = "1.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156"
dependencies = [
"libc",
"rustix 1.1.4",
]
[[package]]
name = "xcursor"
version = "0.3.11"
@@ -6372,10 +6453,13 @@ checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
[[package]]
name = "zerosend"
version = "2.4.1"
version = "2.5.6"
dependencies = [
"arboard",
"async-stream",
"axum",
"blake3",
"bytes",
"chrono",
"clap",
"crossterm",
@@ -6395,6 +6479,7 @@ dependencies = [
"serde_json",
"sha2",
"socket2 0.5.10",
"tar",
"tokio",
"tokio-util",
"tower",
@@ -6402,12 +6487,14 @@ dependencies = [
"tracing",
"tracing-subscriber",
"tray-icon",
"url",
"uuid",
"walkdir",
"windows-sys 0.59.0",
"winreg",
"winres",
"zip",
"zstd",
]
[[package]]
@@ -6458,6 +6545,7 @@ dependencies = [
"memchr",
"thiserror 2.0.20",
"zopfli",
"zstd",
]
[[package]]
@@ -6478,6 +6566,34 @@ dependencies = [
"simd-adler32",
]
[[package]]
name = "zstd"
version = "0.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
dependencies = [
"zstd-safe",
]
[[package]]
name = "zstd-safe"
version = "7.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d"
dependencies = [
"zstd-sys",
]
[[package]]
name = "zstd-sys"
version = "2.0.16+zstd.1.5.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748"
dependencies = [
"cc",
"pkg-config",
]
[[package]]
name = "zune-core"
version = "0.5.3"
+10 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "zerosend"
version = "2.4.1"
version = "2.5.6"
edition = "2021"
description = "Fast, peer-to-peer file transfer designed for ZeroTier, Tailscale, Radmin VPN and LAN"
authors = ["RarDog"]
@@ -10,14 +10,22 @@ authors = ["RarDog"]
tokio = { version = "1.43", features = ["full"] }
tokio-util = { version = "0.7", features = ["io", "codec"] }
futures-util = "0.3"
async-stream = "0.3"
bytes = "1.9"
axum = { version = "0.8", features = ["multipart"] }
tower = "0.5"
tower-http = { version = "0.6", features = ["cors", "trace", "fs"] }
reqwest = { version = "0.12", features = ["stream", "json", "multipart"] }
url = "2.5"
socket2 = { version = "0.5", features = ["all"] }
network-interface = "2.0"
ipnetwork = "0.20"
# Compression & Archiving
zstd = "0.13"
tar = "0.4"
zip = { version = "2.2", default-features = false, features = ["deflate", "zstd"] }
# GUI & Dialogs
eframe = "0.29"
rfd = "0.15"
@@ -25,7 +33,6 @@ qrcode = "0.14"
arboard = "3.4"
image = { version = "0.25", default-features = false, features = ["png"] }
tray-icon = "0.19"
zip = { version = "2.2", default-features = false, features = ["deflate"] }
# CLI & TUI
clap = { version = "4.5", features = ["derive"] }
@@ -42,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]
+42
View File
@@ -41,6 +41,10 @@ pub struct AppConfig {
pub peer_aliases: HashMap<String, String>,
pub window_width: f32,
pub window_height: f32,
#[serde(default)]
pub window_x: Option<f32>,
#[serde(default)]
pub window_y: Option<f32>,
pub close_to_tray: bool,
pub minimize_to_tray: bool,
pub auto_zip_folders: bool,
@@ -59,6 +63,14 @@ pub struct AppConfig {
pub check_updates_on_startup: bool,
#[serde(default)]
pub gitea_api_url: Option<String>,
#[serde(default)]
pub folder_sync_pairs: Vec<crate::transfer::FolderSyncPairConfig>,
#[serde(default = "default_webdav_port")]
pub webdav_port: u16,
}
fn default_webdav_port() -> u16 {
53319
}
fn default_true() -> bool {
@@ -81,6 +93,8 @@ impl Default for AppConfig {
peer_aliases: HashMap::new(),
window_width: 940.0,
window_height: 640.0,
window_x: None,
window_y: None,
close_to_tray: true,
minimize_to_tray: true,
auto_zip_folders: true,
@@ -92,6 +106,8 @@ impl Default for AppConfig {
shared_folders: Vec::new(),
check_updates_on_startup: true,
gitea_api_url: None,
folder_sync_pairs: Vec::new(),
webdav_port: 53319,
}
}
}
@@ -202,3 +218,29 @@ fn get_default_download_dir() -> PathBuf {
let _ = std::fs::create_dir_all(&fallback);
fallback
}
pub fn load_transfers_history() -> Vec<crate::transfer::TransferProgress> {
let path = AppConfig::get_app_dir().join("transfers_history.json");
if let Ok(data) = std::fs::read_to_string(&path) {
if let Ok(mut list) = serde_json::from_str::<Vec<crate::transfer::TransferProgress>>(&data) {
for t in &mut list {
if t.state == crate::transfer::TransferState::InProgress
|| t.state == crate::transfer::TransferState::PendingConfirmation
|| t.state == crate::transfer::TransferState::Paused
{
t.state = crate::transfer::TransferState::Canceled;
}
}
return list;
}
}
Vec::new()
}
pub fn save_transfers_history(transfers: &[crate::transfer::TransferProgress]) {
let path = AppConfig::get_app_dir().join("transfers_history.json");
let to_save: Vec<_> = transfers.iter().take(100).cloned().collect();
if let Ok(json) = serde_json::to_string_pretty(&to_save) {
let _ = std::fs::write(&path, json);
}
}
+32 -7
View File
@@ -32,6 +32,14 @@ fn get_default_device_name() -> String {
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// Clean up any old binaries or temp files from previous self-updates
if let Ok(current_exe) = std::env::current_exe() {
if let Some(exe_dir) = current_exe.parent() {
let _ = std::fs::remove_file(exe_dir.join("zerosend.exe.old"));
let _ = std::fs::remove_file(exe_dir.join("zerosend_update.exe"));
}
}
let config = AppConfig::load();
let cli = Cli::parse();
@@ -184,6 +192,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
TransferState::Rejected => {
pb.abandon_with_message("⛔ Transfer rejected by recipient.");
}
TransferState::Paused => {
pb.set_message("⏸ Paused");
}
TransferState::Canceled => {
pb.abandon_with_message("🛑 Canceled.");
}
@@ -272,6 +283,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
);
server.start().await?;
// Start embedded local WebDAV server for Windows Network Drive mounting
let shared_folders_ref = server.state().shared_folders.clone();
let webdav_port = config.webdav_port;
tokio::spawn(async move {
crate::transfer::start_webdav_server(shared_folders_ref, webdav_port).await;
});
if cli.tui {
ui::tui::run_tui(discovery, server).await?;
} else {
@@ -286,16 +304,23 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
}
let start_minimized = (cli.tray || cli.minimized) && initial_files.is_empty();
let mut viewport = eframe::egui::ViewportBuilder::default()
.with_inner_size([config.window_width, config.window_height])
.with_min_inner_size([780.0, 500.0])
.with_visible(!start_minimized)
.with_title(format!("ZeroSend v{} - P2P Sharing for ZeroTier & VPN", env!("CARGO_PKG_VERSION")));
if let (Some(x), Some(y)) = (config.window_x, config.window_y) {
viewport = viewport.with_position([x, y]);
}
let options = eframe::NativeOptions {
viewport: eframe::egui::ViewportBuilder::default()
.with_inner_size([config.window_width, config.window_height])
.with_min_inner_size([780.0, 500.0])
.with_visible(!start_minimized)
.with_title("ZeroSend v2.4.1 - P2P Sharing for ZeroTier & VPN"),
viewport,
..Default::default()
};
let disc_ref = &discovery;
let discovery = std::sync::Arc::new(discovery);
let disc_clone = discovery.clone();
let srv = server;
let in_req_rx = channels.incoming_request_rx;
let in_txt_rx = channels.incoming_text_rx;
@@ -306,7 +331,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
Box::new(move |cc| {
Ok(Box::new(ui::gui::ZeroSendApp::new(
cc,
disc_ref,
disc_clone,
srv,
config,
initial_files,
+20 -9
View File
@@ -11,6 +11,8 @@ use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
pub const DEFAULT_DISCOVERY_PORT: u16 = 53317;
pub const MULTICAST_IPV4_DISCOVERY: Ipv4Addr = Ipv4Addr::new(239, 255, 255, 250);
pub const MDNS_IPV4_DISCOVERY: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 251);
#[allow(dead_code)]
pub const DEFAULT_TRANSFER_PORT: u16 = 53318;
pub const BEACON_INTERVAL_SECS: u64 = 3;
@@ -143,8 +145,10 @@ impl DiscoveryService {
let (shutdown_tx, _) = tokio::sync::broadcast::channel::<()>(1);
self.shutdown_tx = Some(shutdown_tx.clone());
// 1. Create UDP broadcast sending socket
// 1. Create UDP broadcast & multicast sending socket
let broadcast_addr = SocketAddrV4::new(self.selected_iface.broadcast, DEFAULT_DISCOVERY_PORT);
let multicast_addr = SocketAddrV4::new(MULTICAST_IPV4_DISCOVERY, DEFAULT_DISCOVERY_PORT);
let mdns_addr = SocketAddrV4::new(MDNS_IPV4_DISCOVERY, DEFAULT_DISCOVERY_PORT);
let my_info_static = self.my_info.clone();
let shared_folders = self.shared_folders.clone();
let mut shutdown_rx1 = shutdown_tx.subscribe();
@@ -163,8 +167,8 @@ impl DiscoveryService {
}
info!(
"Started discovery beacon on {} (broadcast: {})",
my_info_static.ip, broadcast_addr
"Started discovery beacon on {} (broadcast: {}, multicast: {})",
my_info_static.ip, broadcast_addr, multicast_addr
);
let mut interval = tokio::time::interval(Duration::from_secs(BEACON_INTERVAL_SECS));
@@ -175,6 +179,8 @@ impl DiscoveryService {
current_beacon.shared_folders = shared_folders.read().await.clone();
if let Ok(data) = serde_json::to_vec(&current_beacon) {
let _ = socket.send_to(&data, broadcast_addr).await;
let _ = socket.send_to(&data, multicast_addr).await;
let _ = socket.send_to(&data, mdns_addr).await;
}
}
_ = shutdown_rx1.recv() => {
@@ -185,15 +191,16 @@ impl DiscoveryService {
}
});
// 2. Create UDP listener socket
let listener_socket = match create_udp_listener(DEFAULT_DISCOVERY_PORT) {
// 2. Create UDP listener socket with Multicast support
let iface_ip = self.selected_iface.ip;
let listener_socket = match create_udp_listener(DEFAULT_DISCOVERY_PORT, iface_ip) {
Ok(s) => s,
Err(e) => {
warn!(
"Could not bind listener on port {}: {}. Will try random port.",
DEFAULT_DISCOVERY_PORT, e
);
create_udp_listener(0)?
create_udp_listener(0, iface_ip)?
}
};
@@ -285,13 +292,13 @@ impl DiscoveryService {
}
}
/// Helper to create a reusable broadcast socket on Windows / Linux / macOS
fn create_udp_listener(port: u16) -> std::io::Result<UdpSocket> {
/// Helper to create a reusable broadcast & multicast socket on Windows / Linux / macOS
fn create_udp_listener(port: u16, iface_ip: Ipv4Addr) -> std::io::Result<UdpSocket> {
let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?;
socket.set_reuse_address(true)?;
#[cfg(not(windows))]
socket.set_reuse_port(true)?;
let _ = socket.set_reuse_port(true);
socket.set_broadcast(true)?;
socket.set_nonblocking(true)?;
@@ -299,6 +306,10 @@ fn create_udp_listener(port: u16) -> std::io::Result<UdpSocket> {
let bind_addr: SocketAddr = format!("0.0.0.0:{}", port).parse().unwrap();
socket.bind(&bind_addr.into())?;
// Join local SSDP/ZeroSend multicast group and mDNS group
let _ = socket.join_multicast_v4(&MULTICAST_IPV4_DISCOVERY, &iface_ip);
let _ = socket.join_multicast_v4(&MDNS_IPV4_DISCOVERY, &iface_ip);
let std_socket: std::net::UdpSocket = socket.into();
UdpSocket::from_std(std_socket)
}
+236
View File
@@ -6,7 +6,12 @@ use walkdir::WalkDir;
use zip::write::SimpleFileOptions;
use zip::{CompressionMethod, ZipArchive, ZipWriter};
// ─────────────────────────────────────────────────────────────────
// Legacy Zip helpers (kept for backward-compat & single-file zips)
// ─────────────────────────────────────────────────────────────────
/// Compress a whole folder into a ZIP archive with fast Deflate compression
#[allow(dead_code)]
pub fn compress_folder_to_zip(
src_dir: &Path,
zip_dest: &Path,
@@ -97,3 +102,234 @@ pub fn extract_zip_archive(
info!("Extraction completed successfully for {:?}", zip_path);
Ok(())
}
// ─────────────────────────────────────────────────────────────────
// Streaming Tar + Zstd (zero disk footprint for folder transfers)
// ─────────────────────────────────────────────────────────────────
/// Returns an estimated uncompressed byte count for the folder
/// (used only to display progress; actual compressed size differs).
pub fn estimate_folder_size(dir: &Path) -> u64 {
WalkDir::new(dir)
.into_iter()
.filter_map(|e| e.ok())
.filter(|e| e.path().is_file())
.map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
.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<Vec<u8>>` channel.
///
/// Run this inside `tokio::task::spawn_blocking` so it does not stall
/// the async executor. The caller receives raw compressed bytes and
/// can forward them directly over the network without any temp file.
pub fn stream_folder_as_tar_zst(
src_dir: PathBuf,
tx: std::sync::mpsc::SyncSender<std::io::Result<Vec<u8>>>,
zstd_level: i32, // 1 = ultra-fast, 3 = balanced
) {
info!("Streaming folder {:?} as tar.zst (level={})", src_dir, zstd_level);
let writer = ChannelWriter::new(tx.clone(), 256 * 1024);
let enc = match zstd::stream::write::Encoder::new(writer, zstd_level) {
Ok(e) => e,
Err(e) => {
let _ = tx.send(Err(std::io::Error::other(e.to_string())));
return;
}
};
let mut tar = tar::Builder::new(enc);
let parent = src_dir.parent().unwrap_or(&src_dir);
for entry in WalkDir::new(&src_dir).into_iter() {
let entry = match entry {
Ok(e) => e,
Err(e) => {
error!("WalkDir error: {}", e);
continue;
}
};
let path = entry.path();
let rel_path = match path.strip_prefix(parent) {
Ok(r) => r,
Err(_) => path,
};
if rel_path.as_os_str().is_empty() || rel_path == Path::new(".") {
continue;
}
if path.is_dir() {
let _ = tar.append_dir(rel_path, path);
} else if path.is_file() {
match File::open(path) {
Ok(mut f) => {
let _ = tar.append_file(rel_path, &mut f);
}
Err(e) => {
error!("Cannot open {:?}: {}", path, e);
}
}
}
}
let enc = match tar.into_inner() {
Ok(e) => e,
Err(e) => {
let _ = tx.send(Err(std::io::Error::other(e.to_string())));
return;
}
};
if let Err(e) = enc.finish() {
let _ = tx.send(Err(std::io::Error::other(e.to_string())));
return;
}
info!("Folder streaming completed for {:?}", src_dir);
}
/// Extract a `tar.zst` byte stream directly into `dest_dir`,
/// with path-traversal protection.
///
/// Call this inside `spawn_blocking`.
pub fn extract_tar_zst_stream<R: Read>(
reader: R,
dest_dir: &Path,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
info!("Extracting tar.zst stream into {:?}", dest_dir);
let decoder = zstd::stream::read::Decoder::new(reader)?;
let mut archive = tar::Archive::new(decoder);
for entry in archive.entries()? {
let mut entry = entry?;
let entry_path = entry.path()?.into_owned();
let safe_rel: PathBuf = entry_path
.components()
.filter(|c| matches!(c, std::path::Component::Normal(_)))
.collect();
if safe_rel.as_os_str().is_empty() {
continue;
}
let outpath = dest_dir.join(&safe_rel);
match entry.header().entry_type() {
tar::EntryType::Directory => {
let _ = std::fs::create_dir_all(&outpath);
}
tar::EntryType::Regular => {
if let Some(p) = outpath.parent() {
let _ = std::fs::create_dir_all(p);
}
if let Ok(mut out) = File::create(&outpath) {
if let Err(e) = std::io::copy(&mut entry, &mut out) {
error!("Write error for {:?}: {}", outpath, e);
}
}
}
_ => {}
}
}
info!("Extraction completed into {:?}", dest_dir);
Ok(())
}
// ─────────────────────────────────────────────────────────────────
// Internal helpers
// ─────────────────────────────────────────────────────────────────
/// A `Write` adapter that accumulates bytes and flushes chunks through
/// a `SyncSender<Result<Vec<u8>>>` when the buffer is full.
struct ChannelWriter {
tx: std::sync::mpsc::SyncSender<std::io::Result<Vec<u8>>>,
buf: Vec<u8>,
chunk_size: usize,
}
impl ChannelWriter {
fn new(
tx: std::sync::mpsc::SyncSender<std::io::Result<Vec<u8>>>,
chunk_size: usize,
) -> Self {
Self {
tx,
buf: Vec::with_capacity(chunk_size),
chunk_size,
}
}
fn flush_buf(&mut self) -> std::io::Result<()> {
if self.buf.is_empty() {
return Ok(());
}
let chunk = std::mem::replace(&mut self.buf, Vec::with_capacity(self.chunk_size));
self.tx
.send(Ok(chunk))
.map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "receiver dropped"))
}
}
impl Write for ChannelWriter {
fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
self.buf.extend_from_slice(data);
if self.buf.len() >= self.chunk_size {
self.flush_buf()?;
}
Ok(data.len())
}
fn flush(&mut self) -> std::io::Result<()> {
self.flush_buf()
}
}
impl Drop for ChannelWriter {
fn drop(&mut self) {
let _ = self.flush_buf();
}
}
+574 -62
View File
@@ -1,15 +1,18 @@
use super::archive::compress_folder_to_zip;
#[allow(unused_imports)]
use super::archive;
use super::protocol::{
FileMetadata, PrepareTransferRequest, PrepareTransferResponse, SharedFolderInfo,
SharedFolderTreeResponse, SharedUploadResponse, TextMessage, TransferProgress, TransferState,
FileMetadata, GrepResponse, PrepareTransferRequest, PrepareTransferResponse,
SharedFileSaveRequest, SharedFileSaveResponse, SharedFolderInfo, SharedFolderTreeResponse,
SharedUploadResponse, TextMessage, TransferProgress, TransferState, WatchPartySyncEvent,
};
use super::rate_limiter::RateLimiter;
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;
@@ -58,6 +61,7 @@ struct ChunkTask {
pub struct TransferClient {
client: Client,
pub pause_flags: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
}
impl TransferClient {
@@ -66,10 +70,63 @@ impl TransferClient {
.timeout(Duration::from_secs(7200))
.connect_timeout(Duration::from_secs(6))
.tcp_nodelay(true)
.tcp_keepalive(Some(Duration::from_secs(15)))
.pool_idle_timeout(Duration::from_secs(120))
.pool_max_idle_per_host(64)
.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);
}
}
#[allow(dead_code)]
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
#[allow(dead_code)]
pub async fn pick_fastest_url(&self, candidates: &[String]) -> Option<String> {
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
@@ -159,8 +216,8 @@ impl TransferClient {
{
let mut file_entries: Vec<(PathBuf, String, u64)> = Vec::new();
let mut total_bytes = 0u64;
let temp_dir = std::env::temp_dir().join("ZeroSend_Zip");
let _ = std::fs::create_dir_all(&temp_dir);
// Folders that need streaming tar.zst transfer (no disk zip)
let mut streaming_folders: Vec<(PathBuf, String)> = Vec::new();
for p in paths {
if !p.exists() {
@@ -174,35 +231,18 @@ impl TransferClient {
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_else(|| "folder".to_string());
let zip_filename = format!("{}.zip", folder_name);
let temp_zip_path = temp_dir.join(&zip_filename);
// Estimate uncompressed size for progress display
let estimated_size = {
let p_clone = p.clone();
tokio::task::spawn_blocking(move || {
super::archive::estimate_folder_size(&p_clone)
})
.await
.unwrap_or(0)
};
// Notify UI that folder is being compressed into ZIP archive
on_progress(TransferProgress {
session_id: Uuid::new_v4().to_string(),
peer_name: peer_display_name.to_string(),
is_incoming: false,
current_file_index: 0,
total_files: paths.len(),
current_file_name: format!("📦 Compressing folder '{}' into ZIP...", folder_name),
bytes_transferred: 0,
total_bytes: 0,
speed_bps: 0,
state: TransferState::InProgress,
checksum_verified: false,
});
let p_clone = p.clone();
let temp_zip_clone = temp_zip_path.clone();
let zip_size = tokio::task::spawn_blocking(move || {
compress_folder_to_zip(&p_clone, &temp_zip_clone)
})
.await
.map_err(|e| format!("Compression task error: {}", e))?
.map_err(|e| format!("Failed to compress folder {:?}: {}", p, e))?;
total_bytes += zip_size;
file_entries.push((temp_zip_path, zip_filename, zip_size));
total_bytes += estimated_size;
streaming_folders.push((p.clone(), folder_name));
} else {
let base_parent = p.parent().unwrap_or(p);
for entry in WalkDir::new(p).into_iter().filter_map(|e| e.ok()) {
@@ -236,10 +276,177 @@ impl TransferClient {
}
}
// ── Stream each folder as tar.zst to the peer (zero disk footprint) ──
for (folder_path, folder_name) in streaming_folders {
let session_id = Uuid::new_v4().to_string();
let session_id_clone = session_id.clone();
let peer_display_name_clone = peer_display_name.to_string();
let folder_name_clone = folder_name.clone();
let folder_est_size = {
let p_c = folder_path.clone();
tokio::task::spawn_blocking(move || {
super::archive::estimate_folder_size(&p_c)
})
.await
.unwrap_or(0)
};
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_display_name.to_string(),
is_incoming: false,
current_file_index: 0,
total_files: paths.len(),
current_file_name: format!("📁 {}", folder_name),
bytes_transferred: 0,
total_bytes: folder_est_size,
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::<std::io::Result<Vec<u8>>>(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, folder_zstd_level);
});
let stream_url = format!(
"{}/api/transfer/folder_stream/{}",
peer_url, session_id
);
// Wrap sync receiver in async stream for reqwest using Arc<Mutex>
let rx_arc = std::sync::Arc::new(std::sync::Mutex::new(rx));
let on_prog_arc = std::sync::Arc::new(std::sync::Mutex::new(on_progress));
let on_prog_stream = on_prog_arc.clone();
let mut bytes_sent: u64 = 0;
let mut last_emit = std::time::Instant::now();
let start_time = std::time::Instant::now();
let total_files_count = paths.len();
let body_stream = async_stream::stream! {
loop {
let rx_clone = rx_arc.clone();
let result = tokio::task::spawn_blocking(move || {
rx_clone.lock().unwrap().recv()
}).await;
match result {
Ok(Ok(Ok(chunk))) => {
let len = chunk.len() as u64;
bytes_sent += len;
if last_emit.elapsed() > std::time::Duration::from_millis(250) {
last_emit = std::time::Instant::now();
let elapsed = start_time.elapsed().as_secs_f64().max(0.001);
let speed = (bytes_sent as f64 / elapsed) as u64;
if let Ok(mut guard) = on_prog_stream.try_lock() {
(guard)(TransferProgress {
session_id: session_id_clone.clone(),
peer_name: peer_display_name_clone.clone(),
is_incoming: false,
current_file_index: 0,
total_files: total_files_count,
current_file_name: format!("📁 {}", folder_name_clone),
bytes_transferred: if folder_est_size > 0 { bytes_sent.min(folder_est_size) } else { bytes_sent },
total_bytes: folder_est_size,
speed_bps: speed,
state: TransferState::InProgress,
checksum_verified: false,
compressed_bytes: bytes_sent,
});
}
}
yield Ok::<_, std::io::Error>(bytes::Bytes::from(chunk));
}
Ok(Ok(Err(e))) => {
yield Err(e);
break;
}
Ok(Err(_)) | Err(_) => break,
}
}
};
let resp = self
.client
.post(&stream_url)
.header("x-folder-name", &folder_name)
.header("x-total-size", folder_est_size.to_string())
.header("x-sender-name", &my_info.name)
.header("x-transfer-type", "tar.zst")
.body(reqwest::Body::wrap_stream(body_stream))
.send()
.await;
compress_handle.await.ok();
on_progress = match std::sync::Arc::try_unwrap(on_prog_arc) {
Ok(mutex) => mutex.into_inner().unwrap(),
Err(arc) => {
let mutex = arc.lock().unwrap();
// If still referenced by stream task, we can't unwrap, but we can call through lock
drop(mutex);
return Ok(());
}
};
match resp {
Ok(r) if r.status().is_success() => {
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_display_name.to_string(),
is_incoming: false,
current_file_index: 0,
total_files: paths.len(),
current_file_name: format!("📁 {}", folder_name),
bytes_transferred: folder_est_size,
total_bytes: folder_est_size,
speed_bps: 0,
state: TransferState::Completed,
checksum_verified: true,
compressed_bytes: bytes_sent,
});
}
Ok(r) => {
return Err(format!(
"Folder stream rejected by peer: HTTP {}",
r.status()
));
}
Err(e) => {
return Err(format!("Failed to stream folder: {}", e));
}
}
}
if file_entries.is_empty() && paths.iter().all(|p| p.is_dir() && auto_zip_folders) {
// All items were streaming folders — done
return Ok(());
}
if file_entries.is_empty() {
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<FileMetadata> = file_entries
.iter()
.map(|(path, rel_path, size)| {
@@ -280,6 +487,7 @@ impl TransferClient {
speed_bps: 0,
state: TransferState::PendingConfirmation,
checksum_verified: false,
compressed_bytes: 0,
});
let prepare_url = format!("{}/api/transfer/prepare", peer_url);
@@ -310,6 +518,7 @@ impl TransferClient {
speed_bps: 0,
state: TransferState::Rejected,
checksum_verified: false,
compressed_bytes: 0,
});
return Err(reason);
@@ -364,9 +573,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::<TransferProgress>();
let s_id = session_id.clone();
@@ -374,15 +590,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();
@@ -413,6 +631,7 @@ impl TransferClient {
speed_bps,
state: TransferState::InProgress,
checksum_verified: false,
compressed_bytes: total_wire,
});
last_time = Instant::now();
@@ -433,6 +652,7 @@ impl TransferClient {
speed_bps: 0,
state: TransferState::InProgress,
checksum_verified: false,
compressed_bytes: already_transferred,
});
let start_time = Instant::now();
@@ -445,6 +665,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 {
@@ -471,16 +692,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(),
@@ -488,9 +734,14 @@ impl TransferClient {
};
if !is_ok {
if chunk_task.retries < 3 {
if chunk_task.retries < 8 {
chunk_task.retries += 1;
tokio::time::sleep(Duration::from_millis(250)).await;
let backoff = (250 * (1 << chunk_task.retries.min(5))).min(4000);
info!(
"Zero-Drop Auto-Recovery: retrying chunk {} (attempt {}/8) after {}ms",
chunk_task.rel_path, chunk_task.retries, backoff
);
tokio::time::sleep(Duration::from_millis(backoff)).await;
tasks_c.lock().await.push_back(chunk_task);
continue;
} else {
@@ -502,7 +753,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>(())
});
@@ -529,6 +780,7 @@ impl TransferClient {
speed_bps: 0,
state: TransferState::Failed(e.clone()),
checksum_verified: false,
compressed_bytes: 0,
});
return Err(e);
}
@@ -549,6 +801,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;
@@ -576,6 +831,7 @@ impl TransferClient {
speed_bps: avg_speed,
state: TransferState::Completed,
checksum_verified: true,
compressed_bytes: total_bytes,
});
Ok(())
@@ -670,37 +926,89 @@ impl TransferClient {
.map_err(|e| format!("Failed to read file text: {}", e))
}
/// Download a file or folder (zip) from a remote shared folder
pub async fn download_shared_file(
/// Download a file or folder (zip) from a remote shared folder with live progress tracking
pub async fn download_shared_file_with_progress<F>(
&self,
peer_url: &str,
folder_id: &str,
subpath: &str,
dest_path: &std::path::Path,
pin: Option<&str>,
) -> Result<u64, String> {
peer_name: &str,
mut on_progress: F,
) -> Result<u64, String>
where
F: FnMut(TransferProgress) + Send + 'static,
{
let clean = subpath.trim_start_matches('/').trim_start_matches('\\');
let mut url = format!("{}/api/shared/download/{}/{}", peer_url, folder_id, clean);
if let Some(p) = pin.filter(|p| !p.trim().is_empty()) {
url = format!("{}?pin={}", url, p);
}
let fname = dest_path.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_else(|| "downloaded".to_string());
let session_id = format!("shared-dl-{}", uuid::Uuid::new_v4());
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_name.to_string(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: fname.clone(),
bytes_transferred: 0,
total_bytes: 0,
speed_bps: 0,
state: TransferState::InProgress,
checksum_verified: false,
compressed_bytes: 0,
});
let resp = self
.client
.get(&url)
.send()
.await
.map_err(|e| format!("Network error: {}", e))?;
.map_err(|e| {
let err_msg = format!("Network error: {}", e);
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_name.to_string(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: fname.clone(),
bytes_transferred: 0,
total_bytes: 0,
speed_bps: 0,
state: TransferState::Failed(err_msg.clone()),
checksum_verified: false,
compressed_bytes: 0,
});
err_msg
})?;
if !resp.status().is_success() {
return Err(format!("Download failed: HTTP {}", resp.status()));
let status = resp.status();
let err_msg = format!("Download failed: HTTP {}", status);
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_name.to_string(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: fname.clone(),
bytes_transferred: 0,
total_bytes: 0,
speed_bps: 0,
state: TransferState::Failed(err_msg.clone()),
checksum_verified: false,
compressed_bytes: 0,
});
return Err(err_msg);
}
let bytes = resp
.bytes()
.await
.map_err(|e| format!("Failed to read stream: {}", e))?;
let len = bytes.len() as u64;
let total_size = resp.content_length().unwrap_or(0);
if let Some(parent) = dest_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
@@ -710,14 +1018,94 @@ impl TransferClient {
.await
.map_err(|e| format!("Could not create file: {}", e))?;
tokio::io::AsyncWriteExt::write_all(&mut file, &bytes)
.await
.map_err(|e| format!("Could not write file: {}", e))?;
use futures_util::StreamExt;
let mut stream = resp.bytes_stream();
let mut bytes_received: u64 = 0;
let start_time = std::time::Instant::now();
let mut last_emit = std::time::Instant::now();
while let Some(chunk_res) = stream.next().await {
match chunk_res {
Ok(chunk) => {
let len = chunk.len() as u64;
bytes_received += len;
tokio::io::AsyncWriteExt::write_all(&mut file, &chunk)
.await
.map_err(|e| format!("Could not write file: {}", e))?;
if last_emit.elapsed() > std::time::Duration::from_millis(250) {
last_emit = std::time::Instant::now();
let elapsed = start_time.elapsed().as_secs_f64().max(0.001);
let speed = (bytes_received as f64 / elapsed) as u64;
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_name.to_string(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: fname.clone(),
bytes_transferred: bytes_received,
total_bytes: if total_size > 0 { total_size } else { bytes_received },
speed_bps: speed,
state: TransferState::InProgress,
checksum_verified: false,
compressed_bytes: 0,
});
}
}
Err(e) => {
let err_msg = format!("Stream error: {}", e);
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_name.to_string(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: fname.clone(),
bytes_transferred: bytes_received,
total_bytes: total_size,
speed_bps: 0,
state: TransferState::Failed(err_msg.clone()),
checksum_verified: false,
compressed_bytes: 0,
});
return Err(err_msg);
}
}
}
tokio::io::AsyncWriteExt::flush(&mut file)
.await
.map_err(|e| format!("Could not flush file: {}", e))?;
Ok(len)
on_progress(TransferProgress {
session_id: session_id.clone(),
peer_name: peer_name.to_string(),
is_incoming: true,
current_file_index: 1,
total_files: 1,
current_file_name: fname.clone(),
bytes_transferred: bytes_received,
total_bytes: if total_size > 0 { total_size } else { bytes_received },
speed_bps: 0,
state: TransferState::Completed,
checksum_verified: true,
compressed_bytes: 0,
});
Ok(bytes_received)
}
/// Download a file or folder (zip) from a remote shared folder
pub async fn download_shared_file(
&self,
peer_url: &str,
folder_id: &str,
subpath: &str,
dest_path: &std::path::Path,
pin: Option<&str>,
) -> Result<u64, String> {
self.download_shared_file_with_progress(peer_url, folder_id, subpath, dest_path, pin, "Shared", |_| {}).await
}
/// Upload a file directly to a friend's shared folder (if Read & Write allowed)
@@ -763,4 +1151,128 @@ impl TransferClient {
.await
.map_err(|e| format!("Failed to parse upload response: {}", e))
}
/// Save updated text/code back to a peer's writable shared folder
pub async fn save_shared_file(
&self,
peer_url: &str,
folder_id: &str,
relative_path: &str,
content: &str,
pin: Option<&str>,
) -> Result<SharedFileSaveResponse, String> {
let clean = relative_path.trim_start_matches('/').trim_start_matches('\\');
let url = format!("{}/api/shared/save/{}/{}", peer_url, folder_id, clean);
let req_body = SharedFileSaveRequest {
content: content.to_string(),
pin: pin.map(|p| p.to_string()),
};
let resp = self
.client
.post(&url)
.json(&req_body)
.send()
.await
.map_err(|e| format!("Save error: {}", e))?;
if !resp.status().is_success() {
return Err(format!("Save failed: HTTP {}", resp.status()));
}
resp.json::<SharedFileSaveResponse>()
.await
.map_err(|e| format!("Failed to parse save response: {}", e))
}
/// Full-text search across remote shared folder
pub async fn grep_shared_folder(
&self,
peer_url: &str,
folder_id: &str,
query: &str,
pin: Option<&str>,
subpath: Option<&str>,
case_sensitive: bool,
) -> Result<GrepResponse, String> {
let mut url = format!("{}/api/shared/grep/{}?query={}", peer_url, folder_id, urlencoding::encode(query));
if let Some(p) = pin.filter(|p| !p.trim().is_empty()) {
url = format!("{}&pin={}", url, urlencoding::encode(p));
}
if let Some(sub) = subpath.filter(|s| !s.trim().is_empty()) {
url = format!("{}&subpath={}", url, urlencoding::encode(sub));
}
if case_sensitive {
url = format!("{}&case_sensitive=true", url);
}
let resp = self
.client
.get(&url)
.send()
.await
.map_err(|e| format!("Search error: {}", e))?;
if !resp.status().is_success() {
return Err(format!("Search failed: HTTP {}", resp.status()));
}
resp.json::<GrepResponse>()
.await
.map_err(|e| format!("Failed to parse search response: {}", e))
}
/// Fetch active watch-party state from host peer
#[allow(dead_code)]
pub async fn get_watch_party(
&self,
peer_url: &str,
) -> Result<Option<WatchPartySyncEvent>, String> {
let url = format!("{}/api/shared/watch-party", peer_url);
let resp = self
.client
.get(&url)
.send()
.await
.map_err(|e| format!("Watch party sync error: {}", e))?;
if !resp.status().is_success() {
return Err(format!("HTTP {}", resp.status()));
}
resp.json::<Option<WatchPartySyncEvent>>()
.await
.map_err(|e| format!("Failed to parse watch party event: {}", e))
}
/// Send watch-party sync event to host peer
#[allow(dead_code)]
pub async fn send_watch_party_event(
&self,
peer_url: &str,
event: &WatchPartySyncEvent,
) -> Result<(), String> {
let url = format!("{}/api/shared/watch-party", peer_url);
let resp = self
.client
.post(&url)
.json(event)
.send()
.await
.map_err(|e| format!("Failed to broadcast watch party event: {}", e))?;
if !resp.status().is_success() {
return Err(format!("Broadcast failed: HTTP {}", resp.status()));
}
Ok(())
}
}
mod urlencoding {
pub fn encode(s: &str) -> String {
url::form_urlencoded::byte_serialize(s.as_bytes()).collect()
}
}
+328
View File
@@ -0,0 +1,328 @@
use crate::transfer::client::TransferClient;
use crate::transfer::protocol::FolderSyncPairConfig;
use chrono::Local;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, RwLock};
use tracing::info;
#[derive(Debug, Clone)]
pub struct SyncStatusUpdate {
#[allow(dead_code)]
pub pair_id: String,
pub message: String,
pub is_error: bool,
pub timestamp: String,
}
pub struct FolderSyncEngine {
client: Arc<TransferClient>,
pairs: Arc<RwLock<Vec<FolderSyncPairConfig>>>,
status_tx: mpsc::UnboundedSender<SyncStatusUpdate>,
}
// ── Sync cache structures ──────────────────────────────────────────────────
#[derive(serde::Serialize, serde::Deserialize, Default)]
struct SyncCache {
files: HashMap<String, CacheEntry>,
}
#[derive(serde::Serialize, serde::Deserialize, Clone)]
struct CacheEntry {
size: u64,
mtime: u64,
blake3: Option<String>,
}
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<String> {
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<TransferClient>,
pairs: Vec<FolderSyncPairConfig>,
status_tx: mpsc::UnboundedSender<SyncStatusUpdate>,
) -> Self {
Self {
client,
pairs: Arc::new(RwLock::new(pairs)),
status_tx,
}
}
#[allow(dead_code)]
pub fn pairs_handle(&self) -> Arc<RwLock<Vec<FolderSyncPairConfig>>> {
self.pairs.clone()
}
pub async fn start_background_sync(
self: Arc<Self>,
peers_map: Arc<RwLock<HashMap<String, crate::network::PeerInfo>>>,
) {
info!("Folder auto-sync engine started.");
loop {
tokio::time::sleep(Duration::from_secs(8)).await;
let pairs = self.pairs.read().await.clone();
for pair in pairs {
if !pair.is_active {
continue;
}
// Check if peer is currently online
let peer_opt = {
let peers = peers_map.read().await;
peers.get(&pair.remote_peer_id).cloned()
};
let peer = match peer_opt {
Some(p) if p.is_online => p,
_ => continue,
};
let peer_url = peer.transfer_url();
let client = self.client.clone();
let status_tx = self.status_tx.clone();
let pair_id = pair.id.clone();
let pair_name = pair.name.clone();
let local_path = pair.local_path.clone();
let remote_folder_id = pair.remote_folder_id.clone();
tokio::spawn(async move {
if let Err(e) = Self::sync_pair(
&client,
&peer_url,
&local_path,
&remote_folder_id,
&pair_id,
&pair_name,
&status_tx,
)
.await
{
let _ = status_tx.send(SyncStatusUpdate {
pair_id,
message: format!("Sync error: {}", e),
is_error: true,
timestamp: Local::now().format("%H:%M:%S").to_string(),
});
}
});
}
}
}
async fn sync_pair(
client: &TransferClient,
peer_url: &str,
local_path: &Path,
remote_folder_id: &str,
pair_id: &str,
pair_name: &str,
status_tx: &mpsc::UnboundedSender<SyncStatusUpdate>,
) -> Result<(), String> {
if !local_path.exists() {
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 with blake3 hash support
// Returns Vec<(rel_path, size, mtime, Option<hash>)>
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<String, (u64, u64, Option<String>)> = 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
let mut remote_files: HashMap<String, u64> = HashMap::new();
for entry in &remote_tree.entries {
if !entry.is_dir {
remote_files.insert(entry.relative_path.clone(), entry.size);
}
}
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, mtime, hash)) in &local_files {
let need_upload = match remote_files.get(rel) {
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
};
if need_upload {
let file_full_path = local_path.join(rel);
let subpath = Path::new(rel).parent().and_then(|p| p.to_str()).unwrap_or("");
if let Ok(_) = client
.upload_to_shared_folder(
peer_url,
remote_folder_id,
subpath,
&file_full_path,
None,
)
.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(),
});
}
}
}
// 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,
None => true, // New remote file
};
if need_download {
let target_file_path = local_path.join(rel);
if let Some(parent) = target_file_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
// Download file
if let Ok(_) = client
.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(),
message: format!("Synced {} item(s) in '{}'", synced_count, pair_name),
is_error: false,
timestamp: Local::now().format("%H:%M:%S").to_string(),
});
}
Ok(())
}
}
/// Walk local directory and return (rel_path, size, mtime, Option<blake3_hash>).
/// Hash is only computed for files < 500MB to avoid stalling.
fn walkdir_local(base: &Path) -> Result<Vec<(String, u64, u64, Option<String>)>, 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() {
if let Ok(rel) = entry.path().strip_prefix(base) {
let rel_str = rel.to_string_lossy().replace('\\', "/");
let meta = entry.metadata().ok();
let size = meta.as_ref().map(|m| m.len()).unwrap_or(0);
let mtime = meta
.as_ref()
.and_then(|m| m.modified().ok())
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs())
.unwrap_or(0);
let hash = if size < MAX_HASH_SIZE {
file_blake3(entry.path())
} else {
None
};
results.push((rel_str, size, mtime, hash));
}
}
}
Ok(results)
}
+4
View File
@@ -1,13 +1,17 @@
pub mod archive;
pub mod client;
pub mod folder_sync;
pub mod protocol;
pub mod rate_limiter;
pub mod server;
pub mod text_analyzer;
pub mod webdav;
pub use client::*;
pub use folder_sync::*;
pub use protocol::*;
#[allow(unused_imports)]
pub use rate_limiter::RateLimiter;
pub use server::*;
pub use text_analyzer::*;
pub use webdav::*;
+59
View File
@@ -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 {
@@ -119,6 +122,62 @@ pub struct SharedUploadResponse {
pub bytes_written: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GrepMatch {
pub file_path: String,
pub line_number: usize,
pub line_content: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GrepResponse {
pub matches: Vec<GrepMatch>,
pub total_files_searched: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SharedFileSaveRequest {
pub content: String,
#[serde(default)]
pub pin: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SharedFileSaveResponse {
pub status: String,
pub bytes_written: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WatchPartySyncEvent {
pub room_id: String,
pub host_peer_id: String,
pub host_peer_name: String,
pub folder_id: String,
pub relative_path: String,
pub is_playing: bool,
pub position_secs: f64,
pub timestamp_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FolderSyncPairConfig {
pub id: String,
pub name: String,
pub local_path: std::path::PathBuf,
pub remote_peer_id: String,
pub remote_peer_name: String,
pub remote_folder_id: String,
#[serde(default = "default_true_val")]
pub is_active: bool,
#[serde(default)]
pub last_sync: Option<String>,
}
fn default_true_val() -> bool {
true
}
pub fn get_mime_type(filename: &str) -> &'static str {
let lower = filename.to_lowercase();
if lower.ends_with(".mp4") {
+656 -10
View File
@@ -1,8 +1,11 @@
#[allow(unused_imports)]
use super::archive;
use super::archive::extract_zip_archive;
use super::protocol::{
FileMetadata, PrepareTransferRequest, PrepareTransferResponse,
SharedFileEntry, SharedFolderInfo, SharedFolderTreeResponse, SharedUploadResponse,
TextMessage, TransferProgress, TransferState,
FileMetadata, GrepMatch, GrepResponse, PrepareTransferRequest, PrepareTransferResponse,
SharedFileEntry, SharedFileSaveRequest, SharedFileSaveResponse, SharedFolderInfo,
SharedFolderTreeResponse, SharedUploadResponse, TextMessage, TransferProgress, TransferState,
WatchPartySyncEvent,
};
use crate::config::SharedFolderConfig;
use crate::network::DiscoveryBeacon;
@@ -36,6 +39,7 @@ pub struct ServerState {
pub pin_code: Arc<RwLock<String>>,
pub auto_extract_zip: Arc<RwLock<bool>>,
pub shared_folders: Arc<RwLock<Vec<SharedFolderConfig>>>,
pub watch_party_state: Arc<RwLock<Option<WatchPartySyncEvent>>>,
pub pending_requests: Arc<RwLock<HashMap<String, PrepareTransferRequest>>>,
pub accepted_requests: Arc<RwLock<HashMap<String, PrepareTransferRequest>>>,
pub pending_confirmations: Arc<Mutex<HashMap<String, oneshot::Sender<bool>>>>,
@@ -85,6 +89,7 @@ impl TransferServer {
pin_code: Arc::new(RwLock::new(pin_code)),
auto_extract_zip: Arc::new(RwLock::new(auto_extract_zip)),
shared_folders: Arc::new(RwLock::new(shared_folders)),
watch_party_state: Arc::new(RwLock::new(None)),
pending_requests: Arc::new(RwLock::new(HashMap::new())),
accepted_requests: Arc::new(RwLock::new(HashMap::new())),
pending_confirmations: Arc::new(Mutex::new(HashMap::new())),
@@ -183,6 +188,10 @@ impl TransferServer {
"/api/transfer/finish/{session_id}",
post(handle_finish_transfer),
)
.route(
"/api/transfer/folder_stream/{session_id}",
post(handle_folder_stream).layer(DefaultBodyLimit::disable()),
)
.route(
"/api/transfer/upload/{session_id}",
post(handle_upload).layer(DefaultBodyLimit::disable()),
@@ -206,12 +215,49 @@ impl TransferServer {
"/api/shared/upload/{folder_id}/{*subpath}",
post(handle_shared_upload).layer(DefaultBodyLimit::disable()),
)
.route("/api/shared/save/{folder_id}/{*subpath}", post(handle_shared_save))
.route("/api/shared/grep/{folder_id}", get(handle_shared_grep))
.route("/api/shared/watch-party", get(handle_watch_party_get).post(handle_watch_party_post))
.layer(CorsLayer::permissive())
.with_state(self.state.clone());
let addr = SocketAddr::from(([0, 0, 0, 0], self.port));
let listener = tokio::net::TcpListener::bind(addr).await?;
info!("ZeroSend HTTP server listening on {}", addr);
let socket = socket2::Socket::new(
socket2::Domain::IPV4,
socket2::Type::STREAM,
Some(socket2::Protocol::TCP),
)?;
socket.set_reuse_address(true)?;
#[cfg(not(windows))]
let _ = socket.set_reuse_port(true);
let _ = socket.set_recv_buffer_size(4 * 1024 * 1024); // 4MB TCP Receive Window
let _ = socket.set_send_buffer_size(4 * 1024 * 1024); // 4MB TCP Send Window
let _ = socket.set_nodelay(true);
socket.set_nonblocking(true)?;
socket.bind(&addr.into())?;
socket.listen(1024)?;
let std_listener: std::net::TcpListener = socket.into();
let listener = tokio::net::TcpListener::from_std(std_listener)?;
info!("ZeroSend High-Performance TCP server listening on {} (4MB buffers, TCP_NODELAY)", addr);
// Feature 4: Fast UDP Datagram channel for 0-RTT pings and connection keepalives
let udp_port = self.port;
let mut udp_shutdown = shutdown_tx.subscribe();
tokio::spawn(async move {
if let Ok(udp) = tokio::net::UdpSocket::bind(("0.0.0.0", udp_port)).await {
let mut buf = [0u8; 128];
loop {
tokio::select! {
Ok((n, src)) = udp.recv_from(&mut buf) => {
if n >= 4 && &buf[..4] == b"PING" {
let _ = udp.send_to(b"PONG", src).await;
}
}
_ = udp_shutdown.recv() => break,
}
}
}
});
let mut shutdown_rx = shutdown_tx.subscribe();
tokio::spawn(async move {
@@ -410,6 +456,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());
@@ -456,6 +503,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());
@@ -539,8 +587,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,
@@ -555,6 +631,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;
@@ -617,6 +694,9 @@ async fn handle_finish_transfer(
let part_path = state.download_dir.join(format!("{}.zerosend_part", safe_rel_path.to_string_lossy()));
if part_path.exists() {
if target_path.exists() {
let _ = tokio::fs::remove_file(&target_path).await;
}
let _ = tokio::fs::rename(&part_path, &target_path).await;
if let Some(expected_hash) = &f.sha256 {
if let Ok(actual_hash) = super::client::compute_file_sha256(&target_path) {
@@ -652,6 +732,179 @@ async fn handle_finish_transfer(
)
}
/// Streaming tar.zst folder receiver — zero disk footprint extraction on-the-fly with live progress
async fn handle_folder_stream(
State(state): State<Arc<ServerState>>,
Path(session_id): Path<String>,
headers: axum::http::HeaderMap,
body: axum::body::Body,
) -> impl IntoResponse {
let raw_folder_name = headers
.get("x-folder-name")
.and_then(|v| v.to_str().ok())
.unwrap_or("folder")
.trim_end_matches(".tar.zst")
.to_string();
let total_size: u64 = headers
.get("x-total-size")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse().ok())
.unwrap_or(0);
let sender_name = headers
.get("x-sender-name")
.and_then(|v| v.to_str().ok())
.unwrap_or("Peer")
.to_string();
info!("Receiving streaming folder: {} (~{} bytes) from {}", raw_folder_name, total_size, sender_name);
// Initial progress event
let initial_progress = super::protocol::TransferProgress {
session_id: session_id.clone(),
peer_name: sender_name.clone(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: format!("📁 {}", raw_folder_name),
bytes_transferred: 0,
total_bytes: total_size,
speed_bps: 0,
state: super::protocol::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);
use futures_util::StreamExt;
let mut stream = body.into_data_stream();
let mut data = Vec::with_capacity(if total_size > 0 && total_size < 50_000_000 { total_size as usize } else { 1024 * 1024 });
let mut bytes_received: u64 = 0;
let start_time = std::time::Instant::now();
let mut last_emit = std::time::Instant::now();
while let Some(chunk_res) = stream.next().await {
match chunk_res {
Ok(chunk) => {
let len = chunk.len() as u64;
bytes_received += len;
data.extend_from_slice(&chunk);
if last_emit.elapsed() > std::time::Duration::from_millis(250) {
last_emit = std::time::Instant::now();
let elapsed = start_time.elapsed().as_secs_f64().max(0.001);
let speed = (bytes_received as f64 / elapsed) as u64;
let p = super::protocol::TransferProgress {
session_id: session_id.clone(),
peer_name: sender_name.clone(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: format!("📁 {}", raw_folder_name),
bytes_transferred: if total_size > 0 { bytes_received.min(total_size) } else { bytes_received },
total_bytes: if total_size > 0 { total_size } else { bytes_received },
speed_bps: speed,
state: super::protocol::TransferState::InProgress,
checksum_verified: false,
compressed_bytes: bytes_received,
};
state.active_transfers.write().await.insert(session_id.clone(), p.clone());
let _ = state.transfer_events.send(p);
}
}
Err(e) => {
error!("Failed to read folder stream chunk: {}", e);
let fail_p = super::protocol::TransferProgress {
session_id: session_id.clone(),
peer_name: sender_name.clone(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: format!("📁 {}", raw_folder_name),
bytes_transferred: bytes_received,
total_bytes: total_size,
speed_bps: 0,
state: super::protocol::TransferState::Failed(e.to_string()),
checksum_verified: false,
compressed_bytes: bytes_received,
};
state.active_transfers.write().await.insert(session_id.clone(), fail_p.clone());
let _ = state.transfer_events.send(fail_p);
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({"error": e.to_string()})),
);
}
}
}
let dest_dir = state.download_dir.clone();
let folder_name_clone = raw_folder_name.clone();
let result = tokio::task::spawn_blocking(move || {
super::archive::extract_tar_zst_stream(std::io::Cursor::new(data), &dest_dir)
})
.await;
match result {
Ok(Ok(())) => {
info!("Folder '{}' extracted successfully", folder_name_clone);
let final_bytes = if total_size > 0 { total_size } else { bytes_received };
let progress = super::protocol::TransferProgress {
session_id: session_id.clone(),
peer_name: sender_name.clone(),
is_incoming: true,
current_file_index: 1,
total_files: 1,
current_file_name: format!("📁 {}", folder_name_clone),
bytes_transferred: final_bytes,
total_bytes: final_bytes,
speed_bps: 0,
state: super::protocol::TransferState::Completed,
checksum_verified: true,
compressed_bytes: bytes_received,
};
state.active_transfers.write().await.insert(session_id.clone(), progress.clone());
let _ = state.transfer_events.send(progress);
(
StatusCode::OK,
Json(serde_json::json!({"status": "extracted", "folder": folder_name_clone})),
)
}
Ok(Err(e)) => {
error!("Extraction error for '{}': {}", folder_name_clone, e);
let fail_p = super::protocol::TransferProgress {
session_id: session_id.clone(),
peer_name: sender_name.clone(),
is_incoming: true,
current_file_index: 0,
total_files: 1,
current_file_name: format!("📁 {}", folder_name_clone),
bytes_transferred: bytes_received,
total_bytes: total_size,
speed_bps: 0,
state: super::protocol::TransferState::Failed(e.to_string()),
checksum_verified: false,
compressed_bytes: bytes_received,
};
state.active_transfers.write().await.insert(session_id.clone(), fail_p.clone());
let _ = state.transfer_events.send(fail_p);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({"error": e.to_string()})),
)
}
Err(e) => {
error!("Spawn blocking error: {}", e);
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({"error": e.to_string()})),
)
}
}
}
async fn handle_upload(
State(state): State<Arc<ServerState>>,
Path(session_id): Path<String>,
@@ -890,6 +1143,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);
@@ -1362,7 +1616,96 @@ async fn handle_shared_download(
let file_size = file.metadata().await.map(|m| m.len()).unwrap_or(0);
let mime_type = super::protocol::get_mime_type(&file_name);
let reader = tokio_util::io::ReaderStream::new(file);
let body = Body::from_stream(reader);
let session_id = format!("shared-dl-{}", Uuid::new_v4());
let peer_label = format!("Общая папка / {}", folder.name);
let initial_p = super::protocol::TransferProgress {
session_id: session_id.clone(),
peer_name: peer_label.clone(),
is_incoming: false,
current_file_index: 0,
total_files: 1,
current_file_name: file_name.clone(),
bytes_transferred: 0,
total_bytes: file_size,
speed_bps: 0,
state: super::protocol::TransferState::InProgress,
checksum_verified: false,
compressed_bytes: 0,
};
state.active_transfers.write().await.insert(session_id.clone(), initial_p.clone());
let _ = state.transfer_events.send(initial_p);
let state_clone = state.clone();
let session_id_clone = session_id.clone();
let peer_label_clone = peer_label.clone();
let file_name_clone = file_name.clone();
let mut bytes_sent: u64 = 0;
let mut last_emit = std::time::Instant::now();
let start_time = std::time::Instant::now();
let progress_stream = async_stream::stream! {
use futures_util::StreamExt;
let mut s = reader;
while let Some(chunk_res) = s.next().await {
match chunk_res {
Ok(chunk) => {
let len = chunk.len() as u64;
bytes_sent += len;
if last_emit.elapsed() > std::time::Duration::from_millis(250) {
last_emit = std::time::Instant::now();
let elapsed = start_time.elapsed().as_secs_f64().max(0.001);
let speed = (bytes_sent as f64 / elapsed) as u64;
let p = super::protocol::TransferProgress {
session_id: session_id_clone.clone(),
peer_name: peer_label_clone.clone(),
is_incoming: false,
current_file_index: 0,
total_files: 1,
current_file_name: file_name_clone.clone(),
bytes_transferred: bytes_sent.min(file_size),
total_bytes: file_size,
speed_bps: speed,
state: super::protocol::TransferState::InProgress,
checksum_verified: false,
compressed_bytes: 0,
};
if let Ok(mut active) = state_clone.active_transfers.try_write() {
active.insert(session_id_clone.clone(), p.clone());
}
let _ = state_clone.transfer_events.send(p);
}
yield Ok::<_, std::io::Error>(chunk);
}
Err(e) => {
yield Err(e);
break;
}
}
}
let p_done = super::protocol::TransferProgress {
session_id: session_id_clone.clone(),
peer_name: peer_label_clone.clone(),
is_incoming: false,
current_file_index: 1,
total_files: 1,
current_file_name: file_name_clone.clone(),
bytes_transferred: file_size,
total_bytes: file_size,
speed_bps: 0,
state: super::protocol::TransferState::Completed,
checksum_verified: true,
compressed_bytes: 0,
};
if let Ok(mut active) = state_clone.active_transfers.try_write() {
active.insert(session_id_clone.clone(), p_done.clone());
}
let _ = state_clone.transfer_events.send(p_done);
};
let body = Body::from_stream(progress_stream);
Response::builder()
.status(StatusCode::OK)
@@ -1418,9 +1761,97 @@ async fn handle_shared_download(
};
let file_size = zip_file.metadata().await.map(|m| m.len()).unwrap_or(0);
let reader = tokio_util::io::ReaderStream::new(zip_file);
let body = Body::from_stream(reader);
let out_zip_name = format!("{}.zip", file_name);
let session_id = format!("shared-dl-{}", Uuid::new_v4());
let peer_label = format!("Общая папка / {}", folder.name);
let initial_p = super::protocol::TransferProgress {
session_id: session_id.clone(),
peer_name: peer_label.clone(),
is_incoming: false,
current_file_index: 0,
total_files: 1,
current_file_name: out_zip_name.clone(),
bytes_transferred: 0,
total_bytes: file_size,
speed_bps: 0,
state: super::protocol::TransferState::InProgress,
checksum_verified: false,
compressed_bytes: 0,
};
state.active_transfers.write().await.insert(session_id.clone(), initial_p.clone());
let _ = state.transfer_events.send(initial_p);
let state_clone = state.clone();
let session_id_clone = session_id.clone();
let peer_label_clone = peer_label.clone();
let out_zip_name_clone = out_zip_name.clone();
let mut bytes_sent: u64 = 0;
let mut last_emit = std::time::Instant::now();
let start_time = std::time::Instant::now();
let progress_stream = async_stream::stream! {
use futures_util::StreamExt;
let mut s = reader;
while let Some(chunk_res) = s.next().await {
match chunk_res {
Ok(chunk) => {
let len = chunk.len() as u64;
bytes_sent += len;
if last_emit.elapsed() > std::time::Duration::from_millis(250) {
last_emit = std::time::Instant::now();
let elapsed = start_time.elapsed().as_secs_f64().max(0.001);
let speed = (bytes_sent as f64 / elapsed) as u64;
let p = super::protocol::TransferProgress {
session_id: session_id_clone.clone(),
peer_name: peer_label_clone.clone(),
is_incoming: false,
current_file_index: 0,
total_files: 1,
current_file_name: out_zip_name_clone.clone(),
bytes_transferred: bytes_sent.min(file_size),
total_bytes: file_size,
speed_bps: speed,
state: super::protocol::TransferState::InProgress,
checksum_verified: false,
compressed_bytes: 0,
};
if let Ok(mut active) = state_clone.active_transfers.try_write() {
active.insert(session_id_clone.clone(), p.clone());
}
let _ = state_clone.transfer_events.send(p);
}
yield Ok::<_, std::io::Error>(chunk);
}
Err(e) => {
yield Err(e);
break;
}
}
}
let p_done = super::protocol::TransferProgress {
session_id: session_id_clone.clone(),
peer_name: peer_label_clone.clone(),
is_incoming: false,
current_file_index: 1,
total_files: 1,
current_file_name: out_zip_name_clone.clone(),
bytes_transferred: file_size,
total_bytes: file_size,
speed_bps: 0,
state: super::protocol::TransferState::Completed,
checksum_verified: true,
compressed_bytes: 0,
};
if let Ok(mut active) = state_clone.active_transfers.try_write() {
active.insert(session_id_clone.clone(), p_done.clone());
}
let _ = state_clone.transfer_events.send(p_done);
};
let body = Body::from_stream(progress_stream);
Response::builder()
.status(StatusCode::OK)
@@ -1521,6 +1952,220 @@ async fn handle_shared_upload(
.into_response()
}
async fn handle_shared_save(
State(state): State<Arc<ServerState>>,
Path((folder_id, subpath)): Path<(String, String)>,
Json(payload): Json<SharedFileSaveRequest>,
) -> Response {
let shares = state.shared_folders.read().await;
let folder = match shares.iter().find(|s| s.id == folder_id) {
Some(f) => f.clone(),
None => {
return (
StatusCode::NOT_FOUND,
Json(serde_json::json!({ "error": "Shared folder not found" })),
)
.into_response();
}
};
drop(shares);
if folder.read_only {
return (
StatusCode::FORBIDDEN,
Json(serde_json::json!({ "error": "Shared folder is Read-Only" })),
)
.into_response();
}
if let Some(expected_pin) = &folder.pin {
if payload.pin.as_deref() != Some(expected_pin.as_str()) {
return (
StatusCode::FORBIDDEN,
Json(serde_json::json!({ "error": "Incorrect or missing PIN code" })),
)
.into_response();
}
}
let base_canonical = match folder.path.canonicalize() {
Ok(p) => p,
Err(_) => {
return (
StatusCode::NOT_FOUND,
Json(serde_json::json!({ "error": "Base path not found" })),
)
.into_response();
}
};
let clean_sub = subpath.trim_start_matches('/').trim_start_matches('\\');
let target_file_path = base_canonical.join(clean_sub);
if !target_file_path.starts_with(&base_canonical) {
return (
StatusCode::FORBIDDEN,
Json(serde_json::json!({ "error": "Path traversal detected" })),
)
.into_response();
}
if let Some(parent) = target_file_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
let bytes = payload.content.as_bytes();
match tokio::fs::write(&target_file_path, bytes).await {
Ok(_) => Json(SharedFileSaveResponse {
status: "success".to_string(),
bytes_written: bytes.len() as u64,
})
.into_response(),
Err(e) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({ "error": format!("Failed to write file: {}", e) })),
)
.into_response(),
}
}
#[derive(serde::Deserialize)]
struct GrepQuery {
query: String,
pin: Option<String>,
subpath: Option<String>,
case_sensitive: Option<bool>,
}
async fn handle_shared_grep(
State(state): State<Arc<ServerState>>,
Path(folder_id): Path<String>,
Query(params): Query<GrepQuery>,
) -> Response {
let shares = state.shared_folders.read().await;
let folder = match shares.iter().find(|s| s.id == folder_id) {
Some(f) => f.clone(),
None => {
return (
StatusCode::NOT_FOUND,
Json(serde_json::json!({ "error": "Shared folder not found" })),
)
.into_response();
}
};
drop(shares);
if let Some(expected_pin) = &folder.pin {
if params.pin.as_deref() != Some(expected_pin.as_str()) {
return (
StatusCode::FORBIDDEN,
Json(serde_json::json!({ "error": "Incorrect or missing PIN code" })),
)
.into_response();
}
}
let base_canonical = match folder.path.canonicalize() {
Ok(p) => p,
Err(_) => {
return (
StatusCode::NOT_FOUND,
Json(serde_json::json!({ "error": "Base path not found" })),
)
.into_response();
}
};
let target_dir = if let Some(sub) = &params.subpath {
let clean = sub.trim_start_matches('/').trim_start_matches('\\');
base_canonical.join(clean)
} else {
base_canonical.clone()
};
if !target_dir.starts_with(&base_canonical) {
return (
StatusCode::FORBIDDEN,
Json(serde_json::json!({ "error": "Path traversal detected" })),
)
.into_response();
}
let search_query = params.query.clone();
let is_case_sens = params.case_sensitive.unwrap_or(false);
let mut matches = Vec::new();
let mut files_searched = 0;
let base_clone = base_canonical.clone();
for entry in walkdir::WalkDir::new(&target_dir).into_iter().filter_map(|e| e.ok()) {
if entry.file_type().is_file() {
let path = entry.path();
let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
let mime = crate::transfer::get_mime_type(name);
let is_text = mime.starts_with("text/")
|| mime.starts_with("application/json")
|| mime.starts_with("application/xml")
|| mime.starts_with("application/javascript")
|| mime.ends_with("x-sh")
|| name.ends_with(".rs")
|| name.ends_with(".py")
|| name.ends_with(".toml")
|| name.ends_with(".md")
|| name.ends_with(".txt")
|| name.ends_with(".log");
if is_text {
files_searched += 1;
if let Ok(content) = std::fs::read_to_string(path) {
if let Ok(rel) = path.strip_prefix(&base_clone) {
let rel_str = rel.to_string_lossy().replace('\\', "/");
for (idx, line) in content.lines().enumerate() {
let is_match = if is_case_sens {
line.contains(&search_query)
} else {
line.to_lowercase().contains(&search_query.to_lowercase())
};
if is_match {
matches.push(GrepMatch {
file_path: rel_str.clone(),
line_number: idx + 1,
line_content: line.trim().to_string(),
});
if matches.len() >= 100 {
break;
}
}
}
}
}
}
}
if matches.len() >= 100 {
break;
}
}
Json(GrepResponse {
matches,
total_files_searched: files_searched,
})
.into_response()
}
async fn handle_watch_party_get(State(state): State<Arc<ServerState>>) -> Response {
let guard = state.watch_party_state.read().await;
Json((*guard).clone()).into_response()
}
async fn handle_watch_party_post(
State(state): State<Arc<ServerState>>,
Json(event): Json<WatchPartySyncEvent>,
) -> Response {
let mut guard = state.watch_party_state.write().await;
*guard = Some(event);
Json(serde_json::json!({ "status": "ok" })).into_response()
}
async fn handle_web_ui(State(state): State<Arc<ServerState>>) -> Html<String> {
let html = format!(
r###"<!DOCTYPE html>
@@ -2110,7 +2755,7 @@ async fn handle_web_ui(State(state): State<Arc<ServerState>>) -> Html<String> {
<div class="toast" id="toast"></div>
<div class="footer">
ZeroSend v2.4.1 • Ultra-fast P2P File & Text Sharing • 🛡️ SHA-256 Verified
ZeroSend v{version} • Ultra-fast P2P File & Text Sharing • 🛡️ SHA-256 Verified
</div>
</div>
@@ -2546,6 +3191,7 @@ async fn handle_web_ui(State(state): State<Arc<ServerState>>) -> Html<String> {
</html>"###,
device_name = state.my_info.name,
net_type = state.my_info.net_type,
version = env!("CARGO_PKG_VERSION"),
);
Html(html)
+431
View File
@@ -0,0 +1,431 @@
use crate::config::SharedFolderConfig;
use axum::{
body::Body,
extract::{Path, State},
http::{header, Method, Request, Response, StatusCode},
routing::any,
Router,
};
use std::sync::Arc;
use tokio::fs::File;
use tokio::sync::RwLock;
use tokio_util::io::ReaderStream;
use tracing::{info, warn};
pub struct WebDavState {
pub shared_folders: Arc<RwLock<Vec<SharedFolderConfig>>>,
#[allow(dead_code)]
pub port: u16,
}
pub fn create_webdav_router(state: Arc<WebDavState>) -> Router {
Router::new()
.route("/webdav", any(handle_webdav_root))
.route("/webdav/{*path}", any(handle_webdav_path))
.with_state(state)
}
pub async fn start_webdav_server(
shared_folders: Arc<RwLock<Vec<SharedFolderConfig>>>,
port: u16,
) {
let state = Arc::new(WebDavState {
shared_folders,
port,
});
let app = create_webdav_router(state);
let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port));
info!("Starting local WebDAV server on http://{}", addr);
match tokio::net::TcpListener::bind(addr).await {
Ok(listener) => {
let _ = axum::serve(listener, app).await;
}
Err(e) => {
warn!("Failed to bind WebDAV server on {}: {}", addr, e);
}
}
}
async fn handle_webdav_root(
State(state): State<Arc<WebDavState>>,
req: Request<Body>,
) -> Response<Body> {
let method = req.method().clone();
let headers = req.headers().clone();
if method == Method::OPTIONS {
return options_response();
}
if method.as_str() == "PROPFIND" {
let shares = state.shared_folders.read().await;
let depth = headers
.get("Depth")
.and_then(|h| h.to_str().ok())
.unwrap_or("1");
let mut xml = String::from("<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<D:multistatus xmlns:D=\"DAV:\">\n");
// Root response
xml.push_str(" <D:response>\n");
xml.push_str(" <D:href>/webdav/</D:href>\n");
xml.push_str(" <D:propstat>\n");
xml.push_str(" <D:prop>\n");
xml.push_str(" <D:displayname>ZeroSend WebDAV</D:displayname>\n");
xml.push_str(" <D:resourcetype><D:collection/></D:resourcetype>\n");
xml.push_str(" </D:prop>\n");
xml.push_str(" <D:status>HTTP/1.1 200 OK</D:status>\n");
xml.push_str(" </D:propstat>\n");
xml.push_str(" </D:response>\n");
if depth != "0" {
for s in shares.iter() {
let encoded_name = urlencoding::encode(&s.name);
xml.push_str(" <D:response>\n");
xml.push_str(&format!(" <D:href>/webdav/{}/</D:href>\n", encoded_name));
xml.push_str(" <D:propstat>\n");
xml.push_str(" <D:prop>\n");
xml.push_str(&format!(" <D:displayname>{}</D:displayname>\n", escape_xml(&s.name)));
xml.push_str(" <D:resourcetype><D:collection/></D:resourcetype>\n");
xml.push_str(" </D:prop>\n");
xml.push_str(" <D:status>HTTP/1.1 200 OK</D:status>\n");
xml.push_str(" </D:propstat>\n");
xml.push_str(" </D:response>\n");
}
}
xml.push_str("</D:multistatus>");
return Response::builder()
.status(StatusCode::MULTI_STATUS)
.header(header::CONTENT_TYPE, "application/xml; charset=utf-8")
.header("DAV", "1, 2")
.header("MS-Author-Via", "DAV")
.body(Body::from(xml))
.unwrap();
}
options_response()
}
async fn handle_webdav_path(
State(state): State<Arc<WebDavState>>,
Path(raw_path): Path<String>,
req: Request<Body>,
) -> Response<Body> {
let method = req.method().clone();
let headers = req.headers().clone();
if method == Method::OPTIONS {
return options_response();
}
let decoded_path = urlencoding::decode(&raw_path)
.unwrap_or(std::borrow::Cow::Borrowed(&raw_path))
.to_string();
let parts: Vec<&str> = decoded_path.split('/').filter(|s| !s.is_empty()).collect();
if parts.is_empty() {
return handle_webdav_root(State(state), req).await;
}
let share_name = parts[0];
let subpath_parts = &parts[1..];
let shares = state.shared_folders.read().await;
let folder = match shares.iter().find(|s| s.name == share_name || s.id == share_name) {
Some(f) => f.clone(),
None => {
return Response::builder()
.status(StatusCode::NOT_FOUND)
.body(Body::from("Folder not found"))
.unwrap();
}
};
drop(shares);
let base_canonical = match folder.path.canonicalize() {
Ok(p) => p,
Err(_) => {
return Response::builder()
.status(StatusCode::NOT_FOUND)
.body(Body::from("Base path not found"))
.unwrap();
}
};
let mut target_path = base_canonical.clone();
for seg in subpath_parts {
if *seg == ".." || *seg == "." {
continue;
}
target_path.push(seg);
}
// Safety check against path traversal
if !target_path.starts_with(&base_canonical) {
return Response::builder()
.status(StatusCode::FORBIDDEN)
.body(Body::from("Forbidden path"))
.unwrap();
}
match method.as_str() {
"PROPFIND" => {
let depth = headers
.get("Depth")
.and_then(|h| h.to_str().ok())
.unwrap_or("1");
let mut xml = String::from("<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<D:multistatus xmlns:D=\"DAV:\">\n");
if let Ok(meta) = std::fs::metadata(&target_path) {
let is_dir = meta.is_dir();
let href_path = format!("/webdav/{}{}", raw_path.trim_end_matches('/'), if is_dir { "/" } else { "" });
xml.push_str(" <D:response>\n");
xml.push_str(&format!(" <D:href>{}</D:href>\n", href_path));
xml.push_str(" <D:propstat>\n");
xml.push_str(" <D:prop>\n");
let name = target_path.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_else(|| share_name.to_string());
xml.push_str(&format!(" <D:displayname>{}</D:displayname>\n", escape_xml(&name)));
if is_dir {
xml.push_str(" <D:resourcetype><D:collection/></D:resourcetype>\n");
} else {
xml.push_str(" <D:resourcetype/>\n");
xml.push_str(&format!(" <D:getcontentlength>{}</D:getcontentlength>\n", meta.len()));
let mime = crate::transfer::get_mime_type(&name);
xml.push_str(&format!(" <D:getcontenttype>{}</D:getcontenttype>\n", mime));
}
xml.push_str(" </D:prop>\n");
xml.push_str(" <D:status>HTTP/1.1 200 OK</D:status>\n");
xml.push_str(" </D:propstat>\n");
xml.push_str(" </D:response>\n");
if is_dir && depth != "0" {
if let Ok(entries) = std::fs::read_dir(&target_path) {
for entry in entries.flatten() {
if let Ok(child_meta) = entry.metadata() {
let child_name = entry.file_name().to_string_lossy().into_owned();
let child_is_dir = child_meta.is_dir();
let child_href = format!("{}/{}{}", href_path.trim_end_matches('/'), urlencoding::encode(&child_name), if child_is_dir { "/" } else { "" });
xml.push_str(" <D:response>\n");
xml.push_str(&format!(" <D:href>{}</D:href>\n", child_href));
xml.push_str(" <D:propstat>\n");
xml.push_str(" <D:prop>\n");
xml.push_str(&format!(" <D:displayname>{}</D:displayname>\n", escape_xml(&child_name)));
if child_is_dir {
xml.push_str(" <D:resourcetype><D:collection/></D:resourcetype>\n");
} else {
xml.push_str(" <D:resourcetype/>\n");
xml.push_str(&format!(" <D:getcontentlength>{}</D:getcontentlength>\n", child_meta.len()));
let mime = crate::transfer::get_mime_type(&child_name);
xml.push_str(&format!(" <D:getcontenttype>{}</D:getcontenttype>\n", mime));
}
xml.push_str(" </D:prop>\n");
xml.push_str(" <D:status>HTTP/1.1 200 OK</D:status>\n");
xml.push_str(" </D:propstat>\n");
xml.push_str(" </D:response>\n");
}
}
}
}
} else {
return Response::builder()
.status(StatusCode::NOT_FOUND)
.body(Body::from("Path not found"))
.unwrap();
}
xml.push_str("</D:multistatus>");
Response::builder()
.status(StatusCode::MULTI_STATUS)
.header(header::CONTENT_TYPE, "application/xml; charset=utf-8")
.header("DAV", "1, 2")
.header("MS-Author-Via", "DAV")
.body(Body::from(xml))
.unwrap()
}
"GET" | "HEAD" => {
if target_path.is_dir() {
return Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "text/html; charset=utf-8")
.body(Body::from("<h1>Directory Index</h1>"))
.unwrap();
}
match File::open(&target_path).await {
Ok(file) => {
let meta = file.metadata().await.ok();
let file_size = meta.as_ref().map(|m| m.len()).unwrap_or(0);
let filename = target_path.file_name().map(|f| f.to_string_lossy().into_owned()).unwrap_or_default();
let mime = crate::transfer::get_mime_type(&filename);
let stream = ReaderStream::new(file);
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, mime)
.header(header::CONTENT_LENGTH, file_size.to_string())
.header(header::ACCEPT_RANGES, "bytes")
.body(Body::from_stream(stream))
.unwrap()
}
Err(_) => Response::builder()
.status(StatusCode::NOT_FOUND)
.body(Body::from("File not found"))
.unwrap(),
}
}
"PUT" => {
if folder.read_only {
return Response::builder()
.status(StatusCode::FORBIDDEN)
.body(Body::from("Folder is read-only"))
.unwrap();
}
let bytes = match axum::body::to_bytes(req.into_body(), 100 * 1024 * 1024).await {
Ok(b) => b,
Err(e) => {
return Response::builder()
.status(StatusCode::BAD_REQUEST)
.body(Body::from(format!("Failed to read body: {}", e)))
.unwrap();
}
};
if let Some(parent) = target_path.parent() {
let _ = std::fs::create_dir_all(parent);
}
if std::fs::write(&target_path, bytes).is_ok() {
Response::builder()
.status(StatusCode::CREATED)
.body(Body::empty())
.unwrap()
} else {
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from("Failed to write file"))
.unwrap()
}
}
"MKCOL" => {
if folder.read_only {
return Response::builder()
.status(StatusCode::FORBIDDEN)
.body(Body::from("Folder is read-only"))
.unwrap();
}
if std::fs::create_dir_all(&target_path).is_ok() {
Response::builder()
.status(StatusCode::CREATED)
.body(Body::empty())
.unwrap()
} else {
Response::builder()
.status(StatusCode::METHOD_NOT_ALLOWED)
.body(Body::from("Failed to create folder"))
.unwrap()
}
}
"DELETE" => {
if folder.read_only {
return Response::builder()
.status(StatusCode::FORBIDDEN)
.body(Body::from("Folder is read-only"))
.unwrap();
}
if target_path.is_dir() {
let _ = std::fs::remove_dir_all(&target_path);
} else {
let _ = std::fs::remove_file(&target_path);
}
Response::builder()
.status(StatusCode::NO_CONTENT)
.body(Body::empty())
.unwrap()
}
_ => options_response(),
}
}
fn options_response() -> Response<Body> {
Response::builder()
.status(StatusCode::OK)
.header("DAV", "1, 2")
.header("MS-Author-Via", "DAV")
.header("Allow", "OPTIONS, GET, HEAD, POST, DELETE, PROPFIND, PROPPATCH, COPY, MOVE, MKCOL, PUT, LOCK, UNLOCK")
.body(Body::empty())
.unwrap()
}
fn escape_xml(s: &str) -> String {
s.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
.replace('\'', "&apos;")
}
mod urlencoding {
pub fn encode(s: &str) -> String {
url::form_urlencoded::byte_serialize(s.as_bytes()).collect()
}
pub fn decode(s: &str) -> Result<std::borrow::Cow<'_, str>, ()> {
url::form_urlencoded::parse(s.as_bytes())
.next()
.map(|(k, _)| k)
.ok_or(())
}
}
/// Helper function to mount Windows Network Drive via `net use`
#[allow(dead_code)]
pub fn mount_webdav_drive(drive_letter: char, port: u16) -> Result<String, String> {
let drive = format!("{}:", drive_letter);
let url = format!("http://127.0.0.1:{}/webdav", port);
// Unmount first if already mapped
let _ = std::process::Command::new("net")
.args(["use", &drive, "/delete", "/y"])
.output();
let output = std::process::Command::new("net")
.args(["use", &drive, &url])
.output()
.map_err(|e| format!("Failed to run net use: {}", e))?;
if output.status.success() {
Ok(format!("Successfully mounted {} to {}", drive, url))
} else {
let err = String::from_utf8_lossy(&output.stderr);
Err(format!("Mount failed: {}", err.trim()))
}
}
/// Helper function to unmount Windows Network Drive
#[allow(dead_code)]
pub fn unmount_webdav_drive(drive_letter: char) -> Result<(), String> {
let drive = format!("{}:", drive_letter);
let output = std::process::Command::new("net")
.args(["use", &drive, "/delete", "/y"])
.output()
.map_err(|e| format!("Failed to run net use: {}", e))?;
if output.status.success() {
Ok(())
} else {
let err = String::from_utf8_lossy(&output.stderr);
Err(format!("Unmount failed: {}", err.trim()))
}
}
+2188 -788
View File
File diff suppressed because it is too large Load Diff
+1
View File
@@ -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(),
};
+326 -114
View File
@@ -9,13 +9,22 @@ pub struct VimViewerModal {
pub file_name: String,
pub relative_path: String,
pub source_peer: String,
pub peer_url: Option<String>,
pub folder_id: Option<String>,
pub pin: Option<String>,
pub is_writable: bool,
pub lines: Vec<String>,
pub buffer_content: String,
pub is_insert_mode: bool,
pub is_markdown_view: bool,
pub cursor_line: usize, // 1-indexed
pub search_query: String,
pub is_searching: bool,
pub command_input: String,
pub is_command_mode: bool,
pub status_msg: Option<String>,
pub pending_save_request: Option<(String, String, String, String, Option<String>)>, // (peer_url, folder_id, rel_path, content, pin)
pub is_dirty: bool,
last_g_press: Option<Instant>,
}
@@ -26,40 +35,113 @@ impl Default for VimViewerModal {
file_name: String::new(),
relative_path: String::new(),
source_peer: String::new(),
peer_url: None,
folder_id: None,
pin: None,
is_writable: false,
lines: Vec::new(),
buffer_content: String::new(),
is_insert_mode: false,
is_markdown_view: false,
cursor_line: 1,
search_query: String::new(),
is_searching: false,
command_input: String::new(),
is_command_mode: false,
status_msg: None,
pending_save_request: None,
is_dirty: false,
last_g_press: None,
}
}
}
impl VimViewerModal {
pub fn open(&mut self, file_name: String, relative_path: String, source_peer: String, content: String) {
#[allow(dead_code)]
pub fn open(
&mut self,
file_name: String,
relative_path: String,
source_peer: String,
content: String,
) {
self.open_with_meta(file_name, relative_path, source_peer, content, None, None, None, false);
}
pub fn open_with_meta(
&mut self,
file_name: String,
relative_path: String,
source_peer: String,
content: String,
peer_url: Option<String>,
folder_id: Option<String>,
pin: Option<String>,
is_writable: bool,
) {
self.is_open = true;
self.file_name = file_name;
self.relative_path = relative_path;
self.source_peer = source_peer;
self.peer_url = peer_url;
self.folder_id = folder_id;
self.pin = pin;
self.is_writable = is_writable;
self.buffer_content = content.clone();
self.lines = content.lines().map(|s| s.to_string()).collect();
if self.lines.is_empty() {
self.lines.push(String::new());
}
self.cursor_line = 1;
self.is_insert_mode = false;
self.is_markdown_view = self.file_name.to_lowercase().ends_with(".md");
self.search_query.clear();
self.is_searching = false;
self.command_input.clear();
self.is_command_mode = false;
self.status_msg = Some(format!("Loaded {} lines", self.lines.len()));
self.is_dirty = false;
self.status_msg = Some(format!(
"Загружено {} строк • {}",
self.lines.len(),
if self.is_writable { "Доступна запись [i: правка, :w: сохранить]" } else { "Только чтение" }
));
self.last_g_press = None;
}
pub fn close(&mut self) {
self.is_open = false;
self.lines.clear();
self.buffer_content.clear();
self.is_insert_mode = false;
self.is_dirty = false;
}
pub fn trigger_save(&mut self) {
if !self.is_writable {
self.status_msg = Some("❌ Невозможно сохранить: папка только для чтения".to_string());
return;
}
if let (Some(url), Some(fid)) = (&self.peer_url, &self.folder_id) {
self.pending_save_request = Some((
url.clone(),
fid.clone(),
self.relative_path.clone(),
self.buffer_content.clone(),
self.pin.clone(),
));
self.status_msg = Some("⏳ Сохранение изменений на удаленном узле...".to_string());
self.is_dirty = false;
} else {
self.status_msg = Some("❌ Нет удаленного подключения для этого файла".to_string());
}
}
pub fn sync_buffer_to_lines(&mut self) {
self.lines = self.buffer_content.lines().map(|s| s.to_string()).collect();
if self.lines.is_empty() {
self.lines.push(String::new());
}
}
pub fn handle_keys(&mut self, ctx: &egui::Context) {
@@ -69,9 +151,13 @@ impl VimViewerModal {
let input = ctx.input(|i| i.clone());
// Escape always cancels search or command mode, or exits if already normal
// Escape always leaves Insert, Search, or Command mode
if input.key_pressed(Key::Escape) {
if self.is_searching {
if self.is_insert_mode {
self.is_insert_mode = false;
self.sync_buffer_to_lines();
self.status_msg = Some("-- NORMAL --".to_string());
} else if self.is_searching {
self.is_searching = false;
} else if self.is_command_mode {
self.is_command_mode = false;
@@ -82,6 +168,11 @@ impl VimViewerModal {
return;
}
if self.is_insert_mode {
// While in insert mode, typing is handled directly by egui::TextEdit
return;
}
if self.is_searching {
if input.key_pressed(Key::Enter) {
self.is_searching = false;
@@ -103,6 +194,17 @@ impl VimViewerModal {
// NORMAL Mode Keybindings
let total = self.lines.len().max(1);
// i or a (enter Insert mode if writable)
if (input.key_pressed(Key::I) || input.key_pressed(Key::A)) && !input.modifiers.ctrl && !input.modifiers.alt {
if self.is_writable {
self.is_insert_mode = true;
self.status_msg = Some("-- INSERT -- (type code, Esc to finish)".to_string());
return;
} else {
self.status_msg = Some("Cannot enter Insert mode: Read-Only share".to_string());
}
}
// j / Down (move down)
if input.key_pressed(Key::J) || input.key_pressed(Key::ArrowDown) {
self.cursor_line = (self.cursor_line + 1).min(total);
@@ -158,14 +260,26 @@ impl VimViewerModal {
}
fn execute_command(&mut self, cmd: &str) {
if cmd == "q" || cmd == "q!" || cmd == "quit" {
self.close();
} else if let Ok(line_num) = cmd.parse::<usize>() {
let total = self.lines.len().max(1);
self.cursor_line = line_num.clamp(1, total);
self.status_msg = Some(format!("Jumped to line {}", self.cursor_line));
} else {
self.status_msg = Some(format!("Unknown command: :{}", cmd));
match cmd {
"w" | "write" => {
self.trigger_save();
}
"wq" | "x" => {
self.trigger_save();
self.close();
}
"q" | "q!" | "quit" => {
self.close();
}
_ => {
if let Ok(line_num) = cmd.parse::<usize>() {
let total = self.lines.len().max(1);
self.cursor_line = line_num.clamp(1, total);
self.status_msg = Some(format!("Jumped to line {}", self.cursor_line));
} else {
self.status_msg = Some(format!("Unknown command: :{}", cmd));
}
}
}
}
@@ -207,54 +321,151 @@ impl VimViewerModal {
let mut is_open = self.is_open;
let screen_rect = ctx.screen_rect();
let win_size = Vec2::new((screen_rect.width() * 0.92).min(1100.0), (screen_rect.height() * 0.88).min(750.0));
let win_size = Vec2::new(
(screen_rect.width() * 0.94).min(1150.0),
(screen_rect.height() * 0.90).min(800.0),
);
egui::Window::new(format!("⚡ Vim Viewer • {} ({})", self.file_name, self.source_peer))
.open(&mut is_open)
.collapsible(false)
.resizable(true)
.default_size(win_size)
.anchor(egui::Align2::CENTER_CENTER, Vec2::ZERO)
.frame(
Frame::window(&ctx.style())
.fill(Color32::from_rgb(15, 18, 28))
.stroke(Stroke::new(1.0_f32, Color32::from_rgb(51, 65, 85)))
.rounding(Rounding::same(8.0))
.inner_margin(Margin::same(0.0)),
)
.show(ctx, |ui| {
// 1. Top Header Bar
Frame::none()
.fill(Color32::from_rgb(22, 27, 44))
.inner_margin(Margin::symmetric(14.0, 10.0))
.show(ui, |ui| {
ui.horizontal(|ui| {
ui.label(RichText::new("📄").size(15.0));
ui.label(
RichText::new(&self.relative_path)
.strong()
.color(Color32::from_rgb(248, 250, 252))
.size(13.0),
);
egui::Window::new(format!(
"⚡ Vim Editor • {} ({})",
self.file_name, self.source_peer
))
.open(&mut is_open)
.collapsible(false)
.resizable(true)
.default_size(win_size)
.anchor(egui::Align2::CENTER_CENTER, Vec2::ZERO)
.frame(
Frame::window(&ctx.style())
.fill(Color32::from_rgb(13, 16, 24))
.stroke(Stroke::new(1.0_f32, Color32::from_rgb(51, 65, 85)))
.rounding(Rounding::same(8.0))
.inner_margin(Margin::same(0.0)),
)
.show(ctx, |ui| {
// 1. Top Header Bar
Frame::none()
.fill(Color32::from_rgb(20, 25, 40))
.inner_margin(Margin::symmetric(14.0, 10.0))
.show(ui, |ui| {
ui.horizontal(|ui| {
ui.label(RichText::new("📄").size(15.0));
ui.label(
RichText::new(&self.relative_path)
.strong()
.color(Color32::from_rgb(248, 250, 252))
.size(13.0),
);
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
if ui.button(RichText::new("✕ Close (:q)").color(Color32::from_rgb(203, 213, 225))).clicked() {
self.is_open = false;
if self.is_dirty {
ui.label(RichText::new("[+]").color(Color32::from_rgb(245, 158, 11)).strong());
}
if self.is_writable {
ui.label(RichText::new("• [Доступна запись]").color(Color32::from_rgb(34, 197, 94)).size(11.0));
} else {
ui.label(RichText::new("• [Только чтение]").color(Color32::from_rgb(148, 163, 184)).size(11.0));
}
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
if ui
.button(RichText::new("✕ Закрыть (:q)").color(Color32::from_rgb(203, 213, 225)))
.clicked()
{
self.close();
}
if self.is_writable {
let save_btn = egui::Button::new(
RichText::new("💾 Сохранить (:w)")
.strong()
.color(Color32::WHITE),
)
.fill(Color32::from_rgb(37, 99, 235))
.rounding(Rounding::same(4.0));
if ui.add(save_btn).clicked() {
self.trigger_save();
}
ui.label(
RichText::new(format!("{} lines", self.lines.len()))
.size(11.0)
.color(Color32::from_rgb(148, 163, 184)),
);
});
let edit_label = if self.is_insert_mode { "👁 Просмотр (Esc)" } else { "✏️ Редактировать (i)" };
if ui.button(RichText::new(edit_label).color(Color32::from_rgb(226, 232, 240))).clicked() {
self.is_insert_mode = !self.is_insert_mode;
if !self.is_insert_mode {
self.sync_buffer_to_lines();
}
}
}
if self.file_name.to_lowercase().ends_with(".md") {
if ui
.selectable_label(self.is_markdown_view, "👁 Предпросмотр Markdown")
.clicked()
{
self.is_markdown_view = !self.is_markdown_view;
}
}
ui.label(
RichText::new(format!("{} строк", self.lines.len()))
.size(11.0)
.color(Color32::from_rgb(148, 163, 184)),
);
});
});
});
// 2. Code Area with Line Numbers Gutter
// 2. Editor / Viewer Body
let available_height = ui.available_height() - 34.0;
if self.is_markdown_view {
// Markdown Preview View
ScrollArea::vertical()
.max_height(available_height)
.auto_shrink([false, false])
.show(ui, |ui| {
Frame::none()
.inner_margin(Margin::same(20.0))
.show(ui, |ui| {
for line in self.buffer_content.lines() {
if line.starts_with("# ") {
ui.heading(RichText::new(&line[2..]).color(Color32::from_rgb(56, 189, 248)).size(20.0));
} else if line.starts_with("## ") {
ui.label(RichText::new(&line[3..]).strong().color(Color32::from_rgb(96, 165, 250)).size(16.0));
} else if line.starts_with("### ") {
ui.label(RichText::new(&line[4..]).strong().color(Color32::from_rgb(147, 197, 253)).size(14.0));
} else if line.starts_with("* ") || line.starts_with("- ") {
ui.label(RichText::new(format!(" • {}", &line[2..])).color(Color32::from_rgb(226, 232, 240)));
} else if line.starts_with("```") {
ui.label(RichText::new(line).monospace().color(Color32::from_rgb(148, 163, 184)));
} else if line.trim().is_empty() {
ui.add_space(6.0);
} else {
ui.label(RichText::new(line).color(Color32::from_rgb(226, 232, 240)));
}
}
});
});
} else if self.is_insert_mode {
// Live Insert Mode Text Editor
ScrollArea::vertical()
.max_height(available_height)
.auto_shrink([false, false])
.show(ui, |ui| {
let edit = TextEdit::multiline(&mut self.buffer_content)
.font(egui::TextStyle::Monospace)
.desired_width(f32::INFINITY)
.lock_focus(true)
.frame(false);
let resp = ui.add(edit);
if resp.changed() {
self.is_dirty = true;
}
});
} else {
// Normal Vim Line-Numbered Viewer
let gutter_width = 54.0;
let available_height = ui.available_height() - 32.0;
ScrollArea::vertical()
.max_height(available_height)
.auto_shrink([false, false])
@@ -266,7 +477,7 @@ impl VimViewerModal {
for idx in 1..=self.lines.len() {
let is_current = idx == self.cursor_line;
let num_color = if is_current {
Color32::from_rgb(245, 158, 11) // Amber current line
Color32::from_rgb(245, 158, 11)
} else {
Color32::from_rgb(71, 85, 105)
};
@@ -281,7 +492,6 @@ impl VimViewerModal {
});
});
// Vertical divider line
ui.add(egui::Separator::default().vertical());
// Code column
@@ -292,7 +502,6 @@ impl VimViewerModal {
for (idx, line) in self.lines.iter().enumerate() {
let line_num = idx + 1;
let is_current = line_num == self.cursor_line;
let line_display = if line.is_empty() { " " } else { line.as_str() };
let line_text = if is_current {
@@ -320,70 +529,73 @@ impl VimViewerModal {
});
});
});
}
// 3. Bottom Vim Status Bar
Frame::none()
.fill(Color32::from_rgb(11, 15, 25))
.stroke(Stroke::new(1.0_f32, Color32::from_rgb(30, 41, 59)))
.inner_margin(Margin::symmetric(10.0, 5.0))
.show(ui, |ui| {
ui.horizontal(|ui| {
// Mode Pill
let (mode_text, mode_bg, mode_fg) = if self.is_searching {
("SEARCH", Color32::from_rgb(217, 119, 6), Color32::WHITE)
} else if self.is_command_mode {
("COMMAND", Color32::from_rgb(147, 51, 234), Color32::WHITE)
} else {
("NORMAL", Color32::from_rgb(37, 99, 235), Color32::WHITE)
};
// 3. Bottom Vim Status Bar
Frame::none()
.fill(Color32::from_rgb(10, 13, 22))
.stroke(Stroke::new(1.0_f32, Color32::from_rgb(30, 41, 59)))
.inner_margin(Margin::symmetric(10.0, 5.0))
.show(ui, |ui| {
ui.horizontal(|ui| {
// Mode Pill
let (mode_text, mode_bg, mode_fg) = if self.is_insert_mode {
("ВСТАВКА", Color32::from_rgb(34, 197, 94), Color32::BLACK)
} else if self.is_searching {
("ПОИСК", Color32::from_rgb(217, 119, 6), Color32::WHITE)
} else if self.is_command_mode {
("КОМАНДА", Color32::from_rgb(147, 51, 234), Color32::WHITE)
} else {
("ОБЫЧНЫЙ", Color32::from_rgb(37, 99, 235), Color32::WHITE)
};
Frame::none()
.fill(mode_bg)
.rounding(Rounding::same(3.0))
.inner_margin(Margin::symmetric(6.0, 2.0))
.show(ui, |ui| {
ui.label(RichText::new(mode_text).size(10.5).strong().color(mode_fg));
});
if self.is_searching {
ui.label(RichText::new("/").monospace().color(Color32::from_rgb(245, 158, 11)));
let edit = TextEdit::singleline(&mut self.search_query)
.hint_text("pattern (press Enter)...")
.font(egui::TextStyle::Monospace)
.desired_width(200.0);
ui.add(edit).request_focus();
} else if self.is_command_mode {
ui.label(RichText::new(":").monospace().color(Color32::from_rgb(168, 85, 247)));
let edit = TextEdit::singleline(&mut self.command_input)
.hint_text("q / line number...")
.font(egui::TextStyle::Monospace)
.desired_width(200.0);
ui.add(edit).request_focus();
} else {
if let Some(msg) = &self.status_msg {
ui.label(RichText::new(msg).size(11.0).color(Color32::from_rgb(203, 213, 225)));
}
}
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
ui.label(
RichText::new("j/k: nav • gg/G: top/bot • /: find • :q: exit")
.size(10.0)
.color(Color32::from_rgb(100, 116, 139)),
);
ui.separator();
ui.label(
RichText::new(format!("Ln {}, Col 1", self.cursor_line))
.size(11.0)
.monospace()
.color(Color32::from_rgb(148, 163, 184)),
);
Frame::none()
.fill(mode_bg)
.rounding(Rounding::same(3.0))
.inner_margin(Margin::symmetric(6.0, 2.0))
.show(ui, |ui| {
ui.label(RichText::new(mode_text).size(10.5).strong().color(mode_fg));
});
if self.is_searching {
ui.label(RichText::new("/").monospace().color(Color32::from_rgb(245, 158, 11)));
let edit = TextEdit::singleline(&mut self.search_query)
.hint_text("поиск (нажмите Enter)...")
.font(egui::TextStyle::Monospace)
.desired_width(200.0);
ui.add(edit).request_focus();
} else if self.is_command_mode {
ui.label(RichText::new(":").monospace().color(Color32::from_rgb(168, 85, 247)));
let edit = TextEdit::singleline(&mut self.command_input)
.hint_text("w (сохранить) / q (выход) / номер...")
.font(egui::TextStyle::Monospace)
.desired_width(220.0);
ui.add(edit).request_focus();
} else {
if let Some(msg) = &self.status_msg {
ui.label(RichText::new(msg).size(11.0).color(Color32::from_rgb(203, 213, 225)));
}
}
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
ui.label(
RichText::new("i: правка • Esc: обычный • :w: сохранить • :q: выход")
.size(10.0)
.color(Color32::from_rgb(100, 116, 139)),
);
ui.separator();
ui.label(
RichText::new(format!("Стр {}, Кол 1", self.cursor_line))
.size(11.0)
.monospace()
.color(Color32::from_rgb(148, 163, 184)),
);
});
});
});
});
});
self.is_open = is_open;
}
+87 -18
View File
@@ -43,6 +43,8 @@ pub struct UpdateInfo {
pub download_url: Option<String>,
pub gitea_url: String,
pub asset_size: u64,
pub is_beta: bool,
pub pre_number: u32,
}
#[derive(Debug, Clone)]
@@ -54,21 +56,72 @@ 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::<u32>().ok()).unwrap_or(0);
let minor = parts.get(1).and_then(|s| s.split('-').next()).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
let patch = parts.get(2).and_then(|s| s.split('-').next()).and_then(|s| s.parse::<u32>().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::<u32>().ok()).unwrap_or(0);
let minor = num_parts.get(1).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
let patch = num_parts.get(2).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
let subpatch = num_parts.get(3).and_then(|s| s.parse::<u32>().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" -> 1001, "beta.1" -> 1000, "beta" -> 0
let pre_num = if let Some(p) = pre_part {
let parts: Vec<&str> = p.split('.').collect();
let p1 = parts.get(1).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
let p2 = parts.get(2).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
if p1 == 0 && p2 == 0 {
parts.last().and_then(|s| s.parse::<u32>().ok()).unwrap_or(0)
} else {
p1 * 1000 + p2
}
} else {
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 +163,15 @@ pub async fn check_gitea_update(api_base: Option<&str>) -> Result<UpdateInfo, St
return Ok(UpdateInfo {
current_version,
latest_version: latest_ver,
latest_version: latest_ver.clone(),
has_update,
release_name: release.name,
release_notes: release.body,
download_url,
gitea_url,
asset_size,
is_beta: is_beta_version(&latest_ver),
pre_number: parse_version(&latest_ver).5,
});
}
}
@@ -149,13 +204,15 @@ pub async fn check_gitea_update(api_base: Option<&str>) -> Result<UpdateInfo, St
return Ok(UpdateInfo {
current_version,
latest_version: latest_ver,
latest_version: latest_ver.clone(),
has_update,
release_name: release.name,
release_notes: release.body,
download_url,
gitea_url,
asset_size,
is_beta: is_beta_version(&latest_ver),
pre_number: parse_version(&latest_ver).5,
});
}
}
@@ -182,6 +239,8 @@ pub async fn check_gitea_update(api_base: Option<&str>) -> Result<UpdateInfo, St
download_url: None,
gitea_url: GITEA_DEFAULT_REPO_URL.to_string(),
asset_size: 0,
is_beta: is_beta_version(parsed),
pre_number: parse_version(parsed).5,
});
}
}
@@ -335,17 +394,27 @@ pub async fn perform_update_download(
info!("Update successfully installed to {:?}", current_exe);
let _ = progress_tx.send(UpdateProgress::Completed);
// Spawn background PowerShell cleanup script to delete old binary once this process terminates
let cleanup_script = format!(
"Start-Sleep -Seconds 1; Remove-Item -Force -ErrorAction SilentlyContinue '{}'",
old_backup_path.to_string_lossy()
);
let _ = std::process::Command::new("powershell")
.args(["-NoProfile", "-WindowStyle", "Hidden", "-Command", &cleanup_script])
.spawn();
// Spawn updated application and exit
let _ = std::process::Command::new(&current_exe).spawn();
std::thread::sleep(Duration::from_millis(500));
std::thread::sleep(Duration::from_millis(400));
std::process::exit(0);
} else {
// Fallback: Use PowerShell deferred replacement
// Fallback: Use PowerShell deferred replacement and cleanup
info!("Direct rename locked; triggering background PowerShell replacement...");
let script = format!(
"Start-Sleep -Milliseconds 600; Move-Item -Force '{temp}' '{target}'; Start-Process '{target}'",
"Start-Sleep -Milliseconds 600; Move-Item -Force '{temp}' '{target}'; Start-Sleep -Seconds 1; Remove-Item -Force -ErrorAction SilentlyContinue '{old}'; Start-Process '{target}'",
temp = temp_download_path.to_string_lossy(),
target = current_exe.to_string_lossy()
target = current_exe.to_string_lossy(),
old = old_backup_path.to_string_lossy()
);
let _ = std::process::Command::new("powershell")
@@ -353,7 +422,7 @@ pub async fn perform_update_download(
.spawn();
let _ = progress_tx.send(UpdateProgress::Completed);
std::thread::sleep(Duration::from_millis(500));
std::thread::sleep(Duration::from_millis(400));
std::process::exit(0);
}
}