1use std::path::{Path, PathBuf};
53use std::sync::{Arc, Mutex};
54
55use futures::StreamExt;
56use libfw_core::capabilities::Capabilities;
57use libfw_core::{HEADER_COMPRESS_LEVEL, HEADER_PROTOCOL, protocol_header_value};
58use reqwest::header::{ACCEPT_ENCODING, AUTHORIZATION, HeaderMap, HeaderValue};
59use serde::Deserialize;
60
61use crate::config::ClientConfig;
62use crate::error::LibfwError;
63use crate::tune::{
64 TuneEvent, TuneParams, TunePhase, TuningEngine, TransferKind, now_ms, resolve_level,
65};
66
67mod download;
68mod upload;
69
70#[derive(Debug, Clone)]
72pub enum NativeEvent {
73 FileStart {
75 path: String,
77 size: u64,
79 },
80 Progress {
82 done: u64,
84 total: u64,
86 },
87 FileDone {
89 path: String,
91 bytes: u64,
93 },
94 Tuning(TuneEvent),
96 Log(String),
98}
99
100pub type EventSink = Arc<dyn Fn(NativeEvent) + Send + Sync>;
102
103#[derive(Debug, Clone)]
105pub struct NativeTuneStatus {
106 pub phase: TunePhase,
108 pub params: TuneParams,
110 pub stats: crate::tune::TuneStats,
112 pub caps_hash: String,
114}
115
116#[derive(Clone)]
118pub struct NativeConfig {
119 pub base_url: String,
121 pub token: String,
123 pub client: ClientConfig,
125 pub on_event: Option<EventSink>,
127}
128
129impl std::fmt::Debug for NativeConfig {
130 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
131 f.debug_struct("NativeConfig")
132 .field("base_url", &self.base_url)
133 .field("token", &"<redacted>")
135 .field("client", &self.client)
136 .field("on_event", &self.on_event.is_some())
137 .finish()
138 }
139}
140
141impl NativeConfig {
142 pub fn new(base_url: impl Into<String>, token: impl Into<String>) -> NativeConfig {
144 NativeConfig {
145 base_url: base_url.into(),
146 token: token.into(),
147 client: ClientConfig::default(),
148 on_event: None,
149 }
150 }
151
152 pub fn with_client(mut self, client: ClientConfig) -> NativeConfig {
154 self.client = client;
155 self
156 }
157
158 pub fn with_events(mut self, sink: EventSink) -> NativeConfig {
160 self.on_event = Some(sink);
161 self
162 }
163}
164
165pub struct NativeClient {
170 cfg: NativeConfig,
171 http: reqwest::Client,
172 tune: Mutex<TuningEngine>,
173}
174
175impl NativeClient {
176 pub fn new(
179 base_url: impl Into<String>,
180 token: impl Into<String>,
181 client: ClientConfig,
182 ) -> NativeClient {
183 NativeClient::with_config(NativeConfig::new(base_url, token).with_client(client))
184 }
185
186 pub fn with_config(cfg: NativeConfig) -> NativeClient {
188 let timeout = std::time::Duration::from_millis(cfg.client.timeout_ms as u64);
189 let http = reqwest::Client::builder()
190 .connect_timeout(timeout)
191 .read_timeout(timeout)
196 .pool_max_idle_per_host(cfg.client.concurrency.max(1) * 2)
197 .build()
198 .expect("reqwest client construction cannot fail");
199 let auto_tune = cfg.client.auto_tune;
200 let compress_level = cfg.client.compress_level;
201 NativeClient {
202 cfg,
203 http,
204 tune: Mutex::new(TuningEngine::new(auto_tune, compress_level)),
205 }
206 }
207
208 pub fn config(&self) -> &ClientConfig {
210 &self.cfg.client
211 }
212
213 pub fn tune_status(&self) -> Option<NativeTuneStatus> {
215 let tune = self.tune.lock().ok()?;
216 if !tune.enabled() {
217 return None;
218 }
219 Some(NativeTuneStatus {
220 phase: tune.phase(),
221 params: tune.params(),
222 stats: tune.stats(),
223 caps_hash: tune.caps_hash().to_string(),
224 })
225 }
226
227 pub fn capabilities(&self) -> Option<Capabilities> {
229 self.tune.lock().ok().and_then(|t| t.caps())
230 }
231
232 pub async fn list(&self, dir: &str) -> Result<Vec<RemoteFile>, LibfwError> {
234 list_dir(self, dir).await
235 }
236
237 pub async fn stat(&self, remote: &str) -> Result<RemoteStat, LibfwError> {
242 validate_relative(remote)?;
243 let (etag, size) = self.fetch_meta(remote).await?;
244 Ok(RemoteStat { size, etag })
245 }
246
247 pub async fn download_folder(
253 &self,
254 dir: &str,
255 dest_dir: impl AsRef<Path>,
256 ) -> Result<u64, LibfwError> {
257 let files = collect_remote_files(self, dir).await?;
258 let total = files.iter().map(|f| f.size).sum();
259 let dest_dir = dest_dir.as_ref().to_path_buf();
260 let concurrency = self.effective_concurrency().max(1);
261
262 let results: Vec<Result<u64, LibfwError>> = futures::stream::iter(files.into_iter().map(
263 |file| {
264 let dest = dest_dir.clone();
265 async move {
266 let local = join_virtual(&dest, &file.path);
268 self.download_file(&file.path, local).await
269 }
270 },
271 ))
272 .buffer_unordered(concurrency)
273 .collect()
274 .await;
275
276 let mut done = 0u64;
277 for result in results {
278 done = done.saturating_add(result?);
279 }
280 self.emit(NativeEvent::Progress { done, total });
281 Ok(done)
282 }
283
284 pub async fn upload_files(
290 &self,
291 files: Vec<(PathBuf, String)>,
292 ) -> Result<u64, LibfwError> {
293 let concurrency = self.effective_concurrency().max(1);
294 let results: Vec<Result<u64, LibfwError>> =
295 futures::stream::iter(files.into_iter().map(|(local, remote)| async move {
296 self.upload_file(&local, &remote).await
297 }))
298 .buffer_unordered(concurrency)
299 .collect()
300 .await;
301
302 let mut done = 0u64;
303 for result in results {
304 done = done.saturating_add(result?);
305 }
306 Ok(done)
307 }
308
309 pub async fn upload_folder(
312 &self,
313 local_dir: impl AsRef<Path>,
314 remote_dir: &str,
315 ) -> Result<u64, LibfwError> {
316 let root = local_dir.as_ref().to_path_buf();
317 let mut files = Vec::new();
318 for entry in walk_files(&root)? {
319 let rel = entry
320 .strip_prefix(&root)
321 .unwrap_or(&entry)
322 .to_string_lossy()
323 .replace('\\', "/");
324 let remote = if remote_dir.is_empty() {
325 rel
326 } else {
327 format!("{}/{}", remote_dir.trim_matches('/'), rel)
328 };
329 files.push((entry, remote));
330 }
331 self.upload_files(files).await
332 }
333
334 pub(crate) fn emit(&self, event: NativeEvent) {
338 if let Some(sink) = &self.cfg.on_event {
339 sink(event);
340 }
341 }
342
343 pub(crate) fn log(&self, msg: impl Into<String>) {
345 self.emit(NativeEvent::Log(msg.into()));
346 }
347
348 pub(crate) fn base_url(&self) -> &str {
349 &self.cfg.base_url
350 }
351
352 pub(crate) fn token(&self) -> &str {
353 &self.cfg.token
354 }
355
356 pub(crate) fn http(&self) -> &reqwest::Client {
357 &self.http
358 }
359
360 pub(crate) fn effective_concurrency(&self) -> usize {
363 match self.tune.lock() {
364 Ok(t) if t.enabled() => t.params().concurrency.max(1),
365 _ => self.cfg.client.concurrency.max(1),
366 }
367 }
368
369 pub(crate) fn with_tune<R>(&self, f: impl FnOnce(&mut TuningEngine) -> R) -> Option<R> {
371 self.tune.lock().ok().map(|mut t| f(&mut t))
372 }
373
374 pub(crate) async fn prepare_transfer(
380 &self,
381 direction: TransferKind,
382 ) -> Result<TuneParams, LibfwError> {
383 let caps = if self.cfg.client.auto_tune {
384 match self.fetch_capabilities().await {
385 Ok(caps) => caps,
386 Err(LibfwError::Http { status: 404, .. }) => {
388 self.with_tune(|t| t.set_enabled(false));
389 Capabilities::default()
390 }
391 Err(e) => return Err(e),
392 }
393 } else {
394 Capabilities::default()
395 };
396
397 let level = resolve_level(self.cfg.client.compress_level, &caps);
398 let static_params = TuneParams::from_config(
399 self.cfg.client.concurrency,
400 self.cfg.client.upload_window,
401 self.cfg.client.download_window,
402 self.cfg.client.chunk_size,
403 level,
404 &caps,
405 );
406 let params = self
407 .with_tune(|t| {
408 t.set_direction(direction);
409 t.begin_transfer(&caps, now_ms(), &static_params)
410 })
411 .unwrap_or(static_params);
412 Ok(params)
413 }
414
415 pub(crate) fn tune_tick(&self, done_bytes: u64, rtt_ms: Option<f64>, error: bool) {
417 if let Some(Some(event)) = self.with_tune(|t| t.tick(now_ms(), done_bytes, rtt_ms, error)) {
418 self.emit(NativeEvent::Tuning(event));
419 }
420 }
421
422 pub(crate) fn finish_transfer(&self, ok: bool) {
425 let _ = self.with_tune(|t| t.transfer_end(ok));
426 }
427
428 pub(crate) fn upload_level(&self, caps: &Capabilities, sample: Option<&[u8]>, mbps: f64) -> i32 {
431 match self.with_tune(|t| t.upload_compress_level(caps, sample, mbps)) {
432 Some(level) => level,
433 None => resolve_level(self.cfg.client.compress_level, caps),
434 }
435 }
436
437 pub async fn fetch_capabilities(&self) -> Result<Capabilities, LibfwError> {
443 let url = format!("{}/capabilities", self.cfg.base_url.trim_end_matches('/'));
444 let resp = self
445 .http
446 .get(&url)
447 .send()
448 .await
449 .map_err(|e| LibfwError::Network(e.to_string()))?;
450 let status = resp.status().as_u16();
451 if status != 200 {
452 return Err(LibfwError::Http { status, url });
453 }
454 let body = resp
455 .bytes()
456 .await
457 .map_err(|e| LibfwError::Network(e.to_string()))?;
458 serde_json::from_slice(&body)
459 .map_err(|e| LibfwError::Protocol(format!("bad /capabilities JSON: {e}")))
460 }
461
462 pub(crate) fn headers(&self, accept_zrip: bool, level: Option<i32>) -> HeaderMap {
465 let mut headers = HeaderMap::new();
466 if let Ok(value) = HeaderValue::from_str(&format!("Bearer {}", self.cfg.token)) {
467 headers.insert(AUTHORIZATION, value);
468 }
469 headers.insert(HEADER_PROTOCOL, HeaderValue::from_static(protocol_header_value()));
470 if accept_zrip {
471 headers.insert(ACCEPT_ENCODING, HeaderValue::from_static("zrip"));
472 if let Some(level) = level {
473 if let Ok(value) = HeaderValue::from_str(&level.to_string()) {
474 headers.insert(HEADER_COMPRESS_LEVEL, value);
475 }
476 }
477 }
478 headers
479 }
480
481 pub(crate) fn file_url(&self, path: &str) -> String {
483 format!(
484 "{}/file/{}",
485 self.cfg.base_url.trim_end_matches('/'),
486 encode_path(path)
487 )
488 }
489
490 pub(crate) fn dir_url(&self, path: &str) -> String {
492 let base = self.cfg.base_url.trim_end_matches('/');
493 if path.is_empty() {
494 format!("{base}/dir")
495 } else {
496 format!("{base}/dir/{}", encode_path(path))
497 }
498 }
499
500 pub(crate) fn backoff(&self, attempt: u32) -> std::time::Duration {
502 std::time::Duration::from_millis(self.cfg.client.backoff_ms(attempt) as u64)
503 }
504}
505
506pub(crate) fn encode_path(path: &str) -> String {
511 let mut out = String::with_capacity(path.len());
512 for byte in path.as_bytes() {
513 match byte {
514 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' | b'/' => {
515 out.push(*byte as char)
516 }
517 other => out.push_str(&format!("%{other:02X}")),
518 }
519 }
520 out
521}
522
523pub(crate) fn join_virtual(base: &Path, virtual_path: &str) -> PathBuf {
525 let mut out = base.to_path_buf();
526 for segment in virtual_path.split('/').filter(|s| !s.is_empty() && *s != ".") {
527 out.push(segment);
528 }
529 out
530}
531
532pub(crate) fn validate_relative(path: &str) -> Result<(), LibfwError> {
534 if path.starts_with('/') {
535 return Err(LibfwError::Protocol(format!(
536 "server returned an absolute path: `{path}`"
537 )));
538 }
539 for segment in path.split('/') {
540 if segment == ".." || segment.contains('\\') {
541 return Err(LibfwError::Protocol(format!(
542 "refusing unsafe path from server: `{path}`"
543 )));
544 }
545 }
546 Ok(())
547}
548
549fn walk_files(root: &Path) -> Result<Vec<PathBuf>, LibfwError> {
551 let mut out = Vec::new();
552 let mut stack = vec![root.to_path_buf()];
553 while let Some(dir) = stack.pop() {
554 let entries = std::fs::read_dir(&dir)
555 .map_err(|e| LibfwError::Storage(format!("read_dir {}: {e}", dir.display())))?;
556 for entry in entries {
557 let entry =
558 entry.map_err(|e| LibfwError::Storage(format!("dir entry: {e}")))?;
559 let file_type = entry
560 .file_type()
561 .map_err(|e| LibfwError::Storage(format!("file type: {e}")))?;
562 let path = entry.path();
563 if file_type.is_symlink() {
564 continue;
566 }
567 if file_type.is_dir() {
568 stack.push(path);
569 } else if file_type.is_file() {
570 out.push(path);
571 }
572 }
573 }
574 out.sort();
575 Ok(out)
576}
577
578#[derive(Debug, Clone, Deserialize)]
580pub struct RemoteFile {
581 pub path: String,
583 #[serde(rename = "is_dir", alias = "isDir")]
585 pub is_dir: bool,
586 #[serde(default)]
588 pub size: u64,
589 #[serde(default)]
591 pub mtime: u64,
592}
593
594impl RemoteFile {
595 pub(crate) fn to_file_entry(&self) -> crate::plan::FileEntry {
597 crate::plan::FileEntry {
598 path: self.path.clone(),
599 size: self.size,
600 mtime: self.mtime,
601 }
602 }
603}
604
605#[derive(Debug, Clone, PartialEq, Eq)]
607pub struct RemoteStat {
608 pub size: u64,
610 pub etag: String,
612}
613
614pub(crate) async fn list_dir(
616 client: &NativeClient,
617 dir: &str,
618) -> Result<Vec<RemoteFile>, LibfwError> {
619 let url = client.dir_url(dir);
620 let resp = client
621 .http()
622 .get(&url)
623 .headers(client.headers(false, None))
624 .send()
625 .await
626 .map_err(|e| LibfwError::Network(e.to_string()))?;
627 let status = resp.status().as_u16();
628 if status != 200 {
629 return Err(LibfwError::Http { status, url });
630 }
631 let body = resp
632 .bytes()
633 .await
634 .map_err(|e| LibfwError::Network(e.to_string()))?;
635 serde_json::from_slice(&body)
636 .map_err(|e| LibfwError::Protocol(format!("bad listing JSON: {e}")))
637}
638
639pub(crate) async fn collect_remote_files(
641 client: &NativeClient,
642 dir: &str,
643) -> Result<Vec<crate::plan::FileEntry>, LibfwError> {
644 let mut out = Vec::new();
645 let mut stack = vec![dir.to_string()];
646 while let Some(current) = stack.pop() {
647 for entry in list_dir(client, ¤t).await? {
648 validate_relative(&entry.path)?;
649 if entry.is_dir {
650 stack.push(entry.path);
651 } else {
652 out.push(entry.to_file_entry());
653 }
654 }
655 }
656 out.sort_by(|a, b| a.path.cmp(&b.path));
657 Ok(out)
658}
659
660pub(crate) fn compress_chunk(raw: &[u8], level: i32) -> Result<Vec<u8>, LibfwError> {
664 use libfw_core::compress::{CompressionFormat, compressor_with_level};
665 let mut enc = compressor_with_level(CompressionFormat::Zrip, level)?;
666 let mut out = Vec::with_capacity(raw.len());
667 for window in raw.chunks(libfw_core::STREAM_BUF_SIZE) {
668 enc.compress(window, &mut out)?;
669 }
670 enc.finish(&mut out)?;
671 Ok(out)
672}
673
674pub(crate) fn session_id_for(meta: &libfw_core::metadata::FileMeta) -> String {
676 meta.etag.trim_matches('"').to_string()
677}
678
679pub(crate) fn is_restart_err(e: &LibfwError) -> bool {
682 matches!(
683 e,
684 LibfwError::Http {
685 status: 416 | 200,
686 ..
687 }
688 )
689}
690
691#[cfg(test)]
692mod tests {
693 use super::*;
694
695 #[test]
696 fn encode_path_preserves_slashes_and_encodes_specials() {
697 assert_eq!(encode_path("a b/c d.txt"), "a%20b/c%20d.txt");
698 assert_eq!(encode_path("simple/file.txt"), "simple/file.txt");
699 assert_eq!(encode_path("dir/中文.pdf"), "dir/%E4%B8%AD%E6%96%87.pdf");
700 assert_eq!(encode_path("a?b#c"), "a%3Fb%23c");
702 }
703
704 #[test]
705 fn join_virtual_builds_nested_paths() {
706 let base = Path::new("/tmp/dest");
707 assert_eq!(
708 join_virtual(base, "a/b/c.txt"),
709 PathBuf::from("/tmp/dest/a/b/c.txt")
710 );
711 assert_eq!(join_virtual(base, ""), PathBuf::from("/tmp/dest"));
712 }
713
714 #[test]
715 fn validate_relative_rejects_traversal() {
716 assert!(validate_relative("a/b.txt").is_ok());
717 assert!(validate_relative("../etc/passwd").is_err());
718 assert!(validate_relative("a/../../b").is_err());
719 assert!(validate_relative("/abs").is_err());
720 assert!(validate_relative("a\\b").is_err());
721 }
722
723 #[test]
724 fn compress_chunk_splits_into_small_frames() {
725 use libfw_core::compress::{CompressionFormat, decompressor};
726 let raw = vec![0x5Au8; 3 * 1024 * 1024];
728 let payload = compress_chunk(&raw, 1).unwrap();
729 assert!(!payload.is_empty());
730 let mut dec = decompressor(CompressionFormat::Zrip);
731 let mut out = Vec::new();
732 dec.decompress(&payload, &mut out).unwrap();
733 dec.finish(&mut out).unwrap();
734 assert_eq!(out, raw);
735 }
736
737 #[test]
738 fn session_id_strips_etag_quotes() {
739 let meta = libfw_core::metadata::FileMeta::new("a.txt", 1, 2);
740 let session = session_id_for(&meta);
741 assert!(!session.contains('"'));
742 assert_eq!(session, meta.etag.trim_matches('"'));
743 }
744}