use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use futures::StreamExt;
use libfw_core::capabilities::Capabilities;
use libfw_core::{HEADER_COMPRESS_LEVEL, HEADER_PROTOCOL, protocol_header_value};
use reqwest::header::{ACCEPT_ENCODING, AUTHORIZATION, HeaderMap, HeaderValue};
use serde::Deserialize;
use crate::config::ClientConfig;
use crate::error::LibfwError;
use crate::tune::{
TuneEvent, TuneParams, TunePhase, TuningEngine, TransferKind, now_ms, resolve_level,
};
mod download;
mod upload;
#[derive(Debug, Clone)]
pub enum NativeEvent {
FileStart {
path: String,
size: u64,
},
Progress {
done: u64,
total: u64,
},
FileDone {
path: String,
bytes: u64,
},
Tuning(TuneEvent),
Log(String),
}
pub type EventSink = Arc<dyn Fn(NativeEvent) + Send + Sync>;
#[derive(Debug, Clone)]
pub struct NativeTuneStatus {
pub phase: TunePhase,
pub params: TuneParams,
pub stats: crate::tune::TuneStats,
pub caps_hash: String,
}
#[derive(Clone)]
pub struct NativeConfig {
pub base_url: String,
pub token: String,
pub client: ClientConfig,
pub on_event: Option<EventSink>,
}
impl std::fmt::Debug for NativeConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("NativeConfig")
.field("base_url", &self.base_url)
.field("token", &"<redacted>")
.field("client", &self.client)
.field("on_event", &self.on_event.is_some())
.finish()
}
}
impl NativeConfig {
pub fn new(base_url: impl Into<String>, token: impl Into<String>) -> NativeConfig {
NativeConfig {
base_url: base_url.into(),
token: token.into(),
client: ClientConfig::default(),
on_event: None,
}
}
pub fn with_client(mut self, client: ClientConfig) -> NativeConfig {
self.client = client;
self
}
pub fn with_events(mut self, sink: EventSink) -> NativeConfig {
self.on_event = Some(sink);
self
}
}
pub struct NativeClient {
cfg: NativeConfig,
http: reqwest::Client,
tune: Mutex<TuningEngine>,
}
impl NativeClient {
pub fn new(
base_url: impl Into<String>,
token: impl Into<String>,
client: ClientConfig,
) -> NativeClient {
NativeClient::with_config(NativeConfig::new(base_url, token).with_client(client))
}
pub fn with_config(cfg: NativeConfig) -> NativeClient {
let timeout = std::time::Duration::from_millis(cfg.client.timeout_ms as u64);
let http = reqwest::Client::builder()
.connect_timeout(timeout)
.read_timeout(timeout)
.pool_max_idle_per_host(cfg.client.concurrency.max(1) * 2)
.build()
.expect("reqwest client construction cannot fail");
let auto_tune = cfg.client.auto_tune;
let compress_level = cfg.client.compress_level;
NativeClient {
cfg,
http,
tune: Mutex::new(TuningEngine::new(auto_tune, compress_level)),
}
}
pub fn config(&self) -> &ClientConfig {
&self.cfg.client
}
pub fn tune_status(&self) -> Option<NativeTuneStatus> {
let tune = self.tune.lock().ok()?;
if !tune.enabled() {
return None;
}
Some(NativeTuneStatus {
phase: tune.phase(),
params: tune.params(),
stats: tune.stats(),
caps_hash: tune.caps_hash().to_string(),
})
}
pub fn capabilities(&self) -> Option<Capabilities> {
self.tune.lock().ok().and_then(|t| t.caps())
}
pub async fn list(&self, dir: &str) -> Result<Vec<RemoteFile>, LibfwError> {
list_dir(self, dir).await
}
pub async fn stat(&self, remote: &str) -> Result<RemoteStat, LibfwError> {
validate_relative(remote)?;
let (etag, size) = self.fetch_meta(remote).await?;
Ok(RemoteStat { size, etag })
}
pub async fn download_folder(
&self,
dir: &str,
dest_dir: impl AsRef<Path>,
) -> Result<u64, LibfwError> {
let files = collect_remote_files(self, dir).await?;
let total = files.iter().map(|f| f.size).sum();
let dest_dir = dest_dir.as_ref().to_path_buf();
let concurrency = self.effective_concurrency().max(1);
let results: Vec<Result<u64, LibfwError>> = futures::stream::iter(files.into_iter().map(
|file| {
let dest = dest_dir.clone();
async move {
let local = join_virtual(&dest, &file.path);
self.download_file(&file.path, local).await
}
},
))
.buffer_unordered(concurrency)
.collect()
.await;
let mut done = 0u64;
for result in results {
done = done.saturating_add(result?);
}
self.emit(NativeEvent::Progress { done, total });
Ok(done)
}
pub async fn upload_files(
&self,
files: Vec<(PathBuf, String)>,
) -> Result<u64, LibfwError> {
let concurrency = self.effective_concurrency().max(1);
let results: Vec<Result<u64, LibfwError>> =
futures::stream::iter(files.into_iter().map(|(local, remote)| async move {
self.upload_file(&local, &remote).await
}))
.buffer_unordered(concurrency)
.collect()
.await;
let mut done = 0u64;
for result in results {
done = done.saturating_add(result?);
}
Ok(done)
}
pub async fn upload_folder(
&self,
local_dir: impl AsRef<Path>,
remote_dir: &str,
) -> Result<u64, LibfwError> {
let root = local_dir.as_ref().to_path_buf();
let mut files = Vec::new();
for entry in walk_files(&root)? {
let rel = entry
.strip_prefix(&root)
.unwrap_or(&entry)
.to_string_lossy()
.replace('\\', "/");
let remote = if remote_dir.is_empty() {
rel
} else {
format!("{}/{}", remote_dir.trim_matches('/'), rel)
};
files.push((entry, remote));
}
self.upload_files(files).await
}
pub(crate) fn emit(&self, event: NativeEvent) {
if let Some(sink) = &self.cfg.on_event {
sink(event);
}
}
pub(crate) fn log(&self, msg: impl Into<String>) {
self.emit(NativeEvent::Log(msg.into()));
}
pub(crate) fn base_url(&self) -> &str {
&self.cfg.base_url
}
pub(crate) fn token(&self) -> &str {
&self.cfg.token
}
pub(crate) fn http(&self) -> &reqwest::Client {
&self.http
}
pub(crate) fn effective_concurrency(&self) -> usize {
match self.tune.lock() {
Ok(t) if t.enabled() => t.params().concurrency.max(1),
_ => self.cfg.client.concurrency.max(1),
}
}
pub(crate) fn with_tune<R>(&self, f: impl FnOnce(&mut TuningEngine) -> R) -> Option<R> {
self.tune.lock().ok().map(|mut t| f(&mut t))
}
pub(crate) async fn prepare_transfer(
&self,
direction: TransferKind,
) -> Result<TuneParams, LibfwError> {
let caps = if self.cfg.client.auto_tune {
match self.fetch_capabilities().await {
Ok(caps) => caps,
Err(LibfwError::Http { status: 404, .. }) => {
self.with_tune(|t| t.set_enabled(false));
Capabilities::default()
}
Err(e) => return Err(e),
}
} else {
Capabilities::default()
};
let level = resolve_level(self.cfg.client.compress_level, &caps);
let static_params = TuneParams::from_config(
self.cfg.client.concurrency,
self.cfg.client.upload_window,
self.cfg.client.download_window,
self.cfg.client.chunk_size,
level,
&caps,
);
let params = self
.with_tune(|t| {
t.set_direction(direction);
t.begin_transfer(&caps, now_ms(), &static_params)
})
.unwrap_or(static_params);
Ok(params)
}
pub(crate) fn tune_tick(&self, done_bytes: u64, rtt_ms: Option<f64>, error: bool) {
if let Some(Some(event)) = self.with_tune(|t| t.tick(now_ms(), done_bytes, rtt_ms, error)) {
self.emit(NativeEvent::Tuning(event));
}
}
pub(crate) fn finish_transfer(&self, ok: bool) {
let _ = self.with_tune(|t| t.transfer_end(ok));
}
pub(crate) fn upload_level(&self, caps: &Capabilities, sample: Option<&[u8]>, mbps: f64) -> i32 {
match self.with_tune(|t| t.upload_compress_level(caps, sample, mbps)) {
Some(level) => level,
None => resolve_level(self.cfg.client.compress_level, caps),
}
}
pub async fn fetch_capabilities(&self) -> Result<Capabilities, LibfwError> {
let url = format!("{}/capabilities", self.cfg.base_url.trim_end_matches('/'));
let resp = self
.http
.get(&url)
.send()
.await
.map_err(|e| LibfwError::Network(e.to_string()))?;
let status = resp.status().as_u16();
if status != 200 {
return Err(LibfwError::Http { status, url });
}
let body = resp
.bytes()
.await
.map_err(|e| LibfwError::Network(e.to_string()))?;
serde_json::from_slice(&body)
.map_err(|e| LibfwError::Protocol(format!("bad /capabilities JSON: {e}")))
}
pub(crate) fn headers(&self, accept_zrip: bool, level: Option<i32>) -> HeaderMap {
let mut headers = HeaderMap::new();
if let Ok(value) = HeaderValue::from_str(&format!("Bearer {}", self.cfg.token)) {
headers.insert(AUTHORIZATION, value);
}
headers.insert(HEADER_PROTOCOL, HeaderValue::from_static(protocol_header_value()));
if accept_zrip {
headers.insert(ACCEPT_ENCODING, HeaderValue::from_static("zrip"));
if let Some(level) = level {
if let Ok(value) = HeaderValue::from_str(&level.to_string()) {
headers.insert(HEADER_COMPRESS_LEVEL, value);
}
}
}
headers
}
pub(crate) fn file_url(&self, path: &str) -> String {
format!(
"{}/file/{}",
self.cfg.base_url.trim_end_matches('/'),
encode_path(path)
)
}
pub(crate) fn dir_url(&self, path: &str) -> String {
let base = self.cfg.base_url.trim_end_matches('/');
if path.is_empty() {
format!("{base}/dir")
} else {
format!("{base}/dir/{}", encode_path(path))
}
}
pub(crate) fn backoff(&self, attempt: u32) -> std::time::Duration {
std::time::Duration::from_millis(self.cfg.client.backoff_ms(attempt) as u64)
}
}
pub(crate) fn encode_path(path: &str) -> String {
let mut out = String::with_capacity(path.len());
for byte in path.as_bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' | b'/' => {
out.push(*byte as char)
}
other => out.push_str(&format!("%{other:02X}")),
}
}
out
}
pub(crate) fn join_virtual(base: &Path, virtual_path: &str) -> PathBuf {
let mut out = base.to_path_buf();
for segment in virtual_path.split('/').filter(|s| !s.is_empty() && *s != ".") {
out.push(segment);
}
out
}
pub(crate) fn validate_relative(path: &str) -> Result<(), LibfwError> {
if path.starts_with('/') {
return Err(LibfwError::Protocol(format!(
"server returned an absolute path: `{path}`"
)));
}
for segment in path.split('/') {
if segment == ".." || segment.contains('\\') {
return Err(LibfwError::Protocol(format!(
"refusing unsafe path from server: `{path}`"
)));
}
}
Ok(())
}
fn walk_files(root: &Path) -> Result<Vec<PathBuf>, LibfwError> {
let mut out = Vec::new();
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
let entries = std::fs::read_dir(&dir)
.map_err(|e| LibfwError::Storage(format!("read_dir {}: {e}", dir.display())))?;
for entry in entries {
let entry =
entry.map_err(|e| LibfwError::Storage(format!("dir entry: {e}")))?;
let file_type = entry
.file_type()
.map_err(|e| LibfwError::Storage(format!("file type: {e}")))?;
let path = entry.path();
if file_type.is_symlink() {
continue;
}
if file_type.is_dir() {
stack.push(path);
} else if file_type.is_file() {
out.push(path);
}
}
}
out.sort();
Ok(out)
}
#[derive(Debug, Clone, Deserialize)]
pub struct RemoteFile {
pub path: String,
#[serde(rename = "is_dir", alias = "isDir")]
pub is_dir: bool,
#[serde(default)]
pub size: u64,
#[serde(default)]
pub mtime: u64,
}
impl RemoteFile {
pub(crate) fn to_file_entry(&self) -> crate::plan::FileEntry {
crate::plan::FileEntry {
path: self.path.clone(),
size: self.size,
mtime: self.mtime,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemoteStat {
pub size: u64,
pub etag: String,
}
pub(crate) async fn list_dir(
client: &NativeClient,
dir: &str,
) -> Result<Vec<RemoteFile>, LibfwError> {
let url = client.dir_url(dir);
let resp = client
.http()
.get(&url)
.headers(client.headers(false, None))
.send()
.await
.map_err(|e| LibfwError::Network(e.to_string()))?;
let status = resp.status().as_u16();
if status != 200 {
return Err(LibfwError::Http { status, url });
}
let body = resp
.bytes()
.await
.map_err(|e| LibfwError::Network(e.to_string()))?;
serde_json::from_slice(&body)
.map_err(|e| LibfwError::Protocol(format!("bad listing JSON: {e}")))
}
pub(crate) async fn collect_remote_files(
client: &NativeClient,
dir: &str,
) -> Result<Vec<crate::plan::FileEntry>, LibfwError> {
let mut out = Vec::new();
let mut stack = vec![dir.to_string()];
while let Some(current) = stack.pop() {
for entry in list_dir(client, ¤t).await? {
validate_relative(&entry.path)?;
if entry.is_dir {
stack.push(entry.path);
} else {
out.push(entry.to_file_entry());
}
}
}
out.sort_by(|a, b| a.path.cmp(&b.path));
Ok(out)
}
pub(crate) fn compress_chunk(raw: &[u8], level: i32) -> Result<Vec<u8>, LibfwError> {
use libfw_core::compress::{CompressionFormat, compressor_with_level};
let mut enc = compressor_with_level(CompressionFormat::Zrip, level)?;
let mut out = Vec::with_capacity(raw.len());
for window in raw.chunks(libfw_core::STREAM_BUF_SIZE) {
enc.compress(window, &mut out)?;
}
enc.finish(&mut out)?;
Ok(out)
}
pub(crate) fn session_id_for(meta: &libfw_core::metadata::FileMeta) -> String {
meta.etag.trim_matches('"').to_string()
}
pub(crate) fn is_restart_err(e: &LibfwError) -> bool {
matches!(
e,
LibfwError::Http {
status: 416 | 200,
..
}
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn encode_path_preserves_slashes_and_encodes_specials() {
assert_eq!(encode_path("a b/c d.txt"), "a%20b/c%20d.txt");
assert_eq!(encode_path("simple/file.txt"), "simple/file.txt");
assert_eq!(encode_path("dir/中文.pdf"), "dir/%E4%B8%AD%E6%96%87.pdf");
assert_eq!(encode_path("a?b#c"), "a%3Fb%23c");
}
#[test]
fn join_virtual_builds_nested_paths() {
let base = Path::new("/tmp/dest");
assert_eq!(
join_virtual(base, "a/b/c.txt"),
PathBuf::from("/tmp/dest/a/b/c.txt")
);
assert_eq!(join_virtual(base, ""), PathBuf::from("/tmp/dest"));
}
#[test]
fn validate_relative_rejects_traversal() {
assert!(validate_relative("a/b.txt").is_ok());
assert!(validate_relative("../etc/passwd").is_err());
assert!(validate_relative("a/../../b").is_err());
assert!(validate_relative("/abs").is_err());
assert!(validate_relative("a\\b").is_err());
}
#[test]
fn compress_chunk_splits_into_small_frames() {
use libfw_core::compress::{CompressionFormat, decompressor};
let raw = vec![0x5Au8; 3 * 1024 * 1024];
let payload = compress_chunk(&raw, 1).unwrap();
assert!(!payload.is_empty());
let mut dec = decompressor(CompressionFormat::Zrip);
let mut out = Vec::new();
dec.decompress(&payload, &mut out).unwrap();
dec.finish(&mut out).unwrap();
assert_eq!(out, raw);
}
#[test]
fn session_id_strips_etag_quotes() {
let meta = libfw_core::metadata::FileMeta::new("a.txt", 1, 2);
let session = session_id_for(&meta);
assert!(!session.contains('"'));
assert_eq!(session, meta.etag.trim_matches('"'));
}
}