tauri-plugin-deeprespond-im 1.0.0-alpha

//! 分片上传与下载（《4.实现与测试》第 3.6 节）。
//!
//! - 证书：rustls 加 rustls-platform-verifier，用系统的证书库校验（与 WebView 一致：企业根证书、私有化部署的自签证书同样可用）；
//! - 代理：Builder 的 `proxy` 优先，否则系统代理（环境变量；Windows、macOS 另读系统设置）；
//! - 上传：从文件按偏移以流的方式读出，每 64 KB 报告一次进度；patches 中的位置在数据流中改写为 `free`，原文件不动；
//! - 下载：先写同目录的 `.part`，完成后改名；本次运行中断过的按 Range 续传（If-Range 保证续上的是同一个文件）。
//!
//! 错误信息不带地址：签名地址中的签名不能进日志。

use std::collections::HashMap;
use std::io::SeekFrom;
use std::path::{Path, PathBuf};
use std::pin::pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use reqwest::header::{HeaderMap, HeaderValue, CONTENT_LENGTH, CONTENT_RANGE, ETAG, IF_RANGE, LAST_MODIFIED, RANGE};
use rustls_platform_verifier::BuilderVerifierExt;
use serde::Deserialize;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
use tokio::sync::Notify;

use super::read::mtime_ms;
use crate::error::{Error, Result};

/// 进度的粒度。
pub const PROGRESS_STEP: u64 = 64 * 1024;
/// 文件名的上限（字节）：给“ (99)”和“.part”留出余地，仍在各系统 255 字节的限制以内。
const MAX_NAME_BYTES: usize = 240;

/// 取消标记：上传、下载在每一块之间和等待网络时检查。
#[derive(Debug, Default)]
pub struct Cancel {
    flag: AtomicBool,
    notify: Notify,
}

impl Cancel {
    pub fn cancel(&self) {
        self.flag.store(true, Ordering::SeqCst);
        self.notify.notify_waiters();
    }

    pub fn is_cancelled(&self) -> bool {
        self.flag.load(Ordering::SeqCst)
    }

    pub async fn cancelled(&self) {
        let mut notified = pin!(self.notify.notified());
        notified.as_mut().enable();
        if self.is_cancelled() {
            return;
        }
        notified.await;
    }
}

/// 上传、下载用的 HTTP 客户端。system_proxy 为 false 时不用任何代理（测试）。
pub fn build_client(proxy: Option<&str>, system_proxy: bool) -> std::result::Result<reqwest::Client, String> {
    let provider = Arc::new(rustls::crypto::ring::default_provider());
    let tls = rustls::ClientConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .map_err(|e| e.to_string())?
        .with_platform_verifier()
        .map_err(|e| format!("系统证书库不可用：{e}"))?
        .with_no_client_auth();
    let mut builder = reqwest::Client::builder()
        .use_preconfigured_tls(tls)
        .connect_timeout(Duration::from_secs(20))
        .read_timeout(Duration::from_secs(60))
        .user_agent(concat!("deeprespond-im-tauri/", env!("CARGO_PKG_VERSION")));
    match proxy {
        Some(url) => {
            let proxy = reqwest::Proxy::all(url).map_err(|e| format!("代理地址无效：{}", e.without_url()))?;
            builder = builder.proxy(proxy.no_proxy(reqwest::NoProxy::from_env()));
        }
        None if !system_proxy => builder = builder.no_proxy(),
        None => {}
    }
    builder.build().map_err(|e| e.without_url().to_string())
}

fn net_error(err: reqwest::Error) -> Error {
    Error::network(err.without_url().to_string())
}

/// 上传前的核对：进入发送队列时记下的大小和修改时间（毫秒）。
#[derive(Debug, Clone, Copy, Deserialize)]
pub struct Expect {
    pub size: u64,
    pub mtime: f64,
}

pub struct UploadPart {
    pub path: PathBuf,
    pub offset: u64,
    pub length: u64,
    pub url: String,
    pub method: String,
    pub headers: HashMap<String, String>,
    pub expect: Option<Expect>,
    pub patches: Vec<u64>,
}

fn changed() -> Error {
    Error::new("send_abandoned", Some("attachment_changed"), "附件在发送前被修改")
}

/// 把 [pos, pos + buf.len()) 中落在 patches（4 字节的类型字段）里的字节改写为 free。
pub fn apply_patches(buf: &mut [u8], pos: u64, patches: &[u64]) {
    let end = pos + buf.len() as u64;
    for &p in patches {
        for (k, byte) in b"free".iter().enumerate() {
            let at = p + k as u64;
            if at >= pos && at < end {
                buf[(at - pos) as usize] = *byte;
            }
        }
    }
}

struct Reader {
    file: tokio::fs::File,
    pos: u64,
    remaining: u64,
    loaded: u64,
    done: bool,
}

/// 上传一片：返回 HTTP 状态码（是否成功由 JavaScript 判断，与 Web SDK 的 putObject 相同）。
pub async fn upload_part(client: &reqwest::Client, part: UploadPart, cancel: Arc<Cancel>, progress: Arc<dyn Fn(u64) + Send + Sync>) -> Result<u16> {
    let meta = tokio::fs::metadata(&part.path).await.map_err(|e| Error::from_io_read(&e))?;
    if let Some(expect) = part.expect {
        if meta.len() != expect.size || mtime_ms(&meta) as i64 != expect.mtime.round() as i64 {
            return Err(changed());
        }
    }
    if part.offset.checked_add(part.length).is_none_or(|end| end > meta.len()) {
        return Err(Error::validation("file_unreadable", "文件比要上传的范围短"));
    }
    let method = reqwest::Method::from_bytes(part.method.as_bytes()).map_err(|_| Error::validation("invalid_method", "上传的方法无效"))?;
    let mut file = tokio::fs::File::open(&part.path).await.map_err(|e| Error::from_io_read(&e))?;
    file.seek(SeekFrom::Start(part.offset)).await.map_err(|e| Error::from_io_read(&e))?;
    let failure: Arc<Mutex<Option<Error>>> = Arc::default();
    let patches = Arc::new(part.patches);
    let reader = Reader { file, pos: part.offset, remaining: part.length, loaded: 0, done: false };
    let stream = {
        let (failure, cancel, patches, progress) = (failure.clone(), cancel.clone(), patches.clone(), progress.clone());
        futures_util::stream::unfold(reader, move |mut st| {
            let (failure, cancel, patches, progress) = (failure.clone(), cancel.clone(), patches.clone(), progress.clone());
            async move {
                if st.done || st.remaining == 0 {
                    return None;
                }
                if cancel.is_cancelled() {
                    st.done = true;
                    return Some((Err(std::io::Error::other("aborted")), st));
                }
                let n = st.remaining.min(PROGRESS_STEP) as usize;
                let mut buf = vec![0u8; n];
                if let Err(err) = st.file.read_exact(&mut buf).await {
                    *failure.lock().unwrap() = Some(Error::from_io_read(&err));
                    st.done = true;
                    return Some((Err(err), st));
                }
                apply_patches(&mut buf, st.pos, &patches);
                st.pos += n as u64;
                st.remaining -= n as u64;
                st.loaded += n as u64;
                progress(st.loaded);
                Some((Ok::<_, std::io::Error>(buf), st))
            }
        })
    };
    // 对象存储的预签名 PUT 不接受分块编码：明确给出长度
    let mut request = client.request(method, &part.url).header(CONTENT_LENGTH, part.length).body(reqwest::Body::wrap_stream(stream));
    for (key, value) in &part.headers {
        if key.eq_ignore_ascii_case("content-length") || key.eq_ignore_ascii_case("host") {
            continue;
        }
        request = request.header(key.as_str(), value.as_str());
    }
    let result = tokio::select! {
        result = request.send() => result,
        _ = cancel.cancelled() => return Err(Error::aborted()),
    };
    match result {
        Ok(response) => Ok(response.status().as_u16()),
        Err(err) => {
            if let Some(local) = failure.lock().unwrap().take() {
                return Err(local);
            }
            if cancel.is_cancelled() {
                return Err(Error::aborted());
            }
            Err(net_error(err))
        }
    }
}

/// 本次运行中写过的 `.part`：只续传自己写的（上次运行或别的下载留下的同名 .part 可能是另一个文件），
/// 并记下 ETag 或 Last-Modified，续传时以 If-Range 确认服务端的文件没有变。
#[derive(Default)]
pub struct Parts {
    known: Mutex<HashMap<PathBuf, Validator>>,
}

#[derive(Clone, Default)]
struct Validator {
    etag: Option<String>,
    last_modified: Option<String>,
}

pub fn part_path(target: &Path) -> PathBuf {
    let mut name = target.file_name().map(|n| n.to_os_string()).unwrap_or_default();
    name.push(".part");
    target.with_file_name(name)
}

/// 已有的路径（含悬空的符号链接）。
pub fn exists(path: &Path) -> bool {
    std::fs::symlink_metadata(path).is_ok()
}

/// 下载到 target：返回最终的路径和大小。rename 为 true 时，完成时 target 已被别的文件占用的改用下一个名称。
pub async fn download(client: &reqwest::Client, url: &str, target: &Path, rename: bool, parts: &Parts, cancel: &Cancel, progress: &(dyn Fn(u64, Option<u64>) + Send + Sync)) -> Result<(PathBuf, u64)> {
    let part = part_path(target);
    if std::fs::symlink_metadata(&part).is_ok_and(|m| m.file_type().is_symlink()) {
        return Err(Error::permission_denied("path_not_allowed"));
    }
    let result = fetch(client, url, &part, parts, cancel, progress).await;
    let size = match result {
        Ok(size) => size,
        Err(err) => {
            // 取消的不再续传：删除 .part；网络中断、地址过期的留着，下次续传
            if err.code == "aborted" {
                parts.known.lock().unwrap().remove(&part);
                let _ = tokio::fs::remove_file(&part).await;
            }
            return Err(err);
        }
    };
    parts.known.lock().unwrap().remove(&part);
    let dest = if rename && exists(target) { next_free(target, |_| false) } else { target.to_path_buf() };
    tokio::fs::rename(&part, &dest).await.map_err(|e| Error::from_io(&e))?;
    Ok((dest, size))
}

async fn fetch(client: &reqwest::Client, url: &str, part: &Path, parts: &Parts, cancel: &Cancel, progress: &(dyn Fn(u64, Option<u64>) + Send + Sync)) -> Result<u64> {
    for _attempt in 0..2 {
        let known = parts.known.lock().unwrap().get(part).cloned();
        let mut offset = match (&known, tokio::fs::metadata(part).await) {
            (Some(_), Ok(meta)) if meta.is_file() => meta.len(),
            _ => 0,
        };
        let mut headers = HeaderMap::new();
        if offset > 0 {
            headers.insert(RANGE, HeaderValue::from_str(&format!("bytes={offset}-")).unwrap());
            let validator = known.as_ref().and_then(|v| v.etag.clone().filter(|e| !e.starts_with("W/")).or_else(|| v.last_modified.clone()));
            if let Some(value) = validator.and_then(|v| HeaderValue::from_str(&v).ok()) {
                headers.insert(IF_RANGE, value);
            }
        }
        let response = tokio::select! {
            r = client.get(url).headers(headers).send() => r.map_err(net_error)?,
            _ = cancel.cancelled() => return Err(Error::aborted()),
        };
        let status = response.status().as_u16();
        let content_range = response.headers().get(CONTENT_RANGE).and_then(|v| v.to_str().ok()).map(str::to_string);
        let total = match status {
            403 => return Err(Error::new("url_expired", None, "下载地址已过期或无权访问")),
            416 if offset > 0 => {
                // 已经下载完整（.part 就是整个文件）；否则从头再来
                if content_range.as_deref().and_then(parse_unsatisfied) == Some(offset) {
                    return Ok(offset);
                }
                parts.known.lock().unwrap().remove(part);
                continue;
            }
            206 => match content_range.as_deref().and_then(parse_content_range) {
                Some((start, total)) if start == offset => total,
                _ => {
                    parts.known.lock().unwrap().remove(part);
                    continue;
                }
            },
            200 => {
                // 服务端不支持 Range 或文件变了（If-Range 不符）：从头写
                offset = 0;
                response.content_length()
            }
            s if (200..300).contains(&s) => {
                offset = 0;
                response.content_length()
            }
            404 => return Err(Error::new("not_found", None, format!("下载失败：{status}"))),
            s if s >= 500 => return Err(Error::new("internal", None, format!("下载失败：{status}"))),
            _ => return Err(Error::new("download_failed", None, format!("下载失败：{status}"))),
        };
        let validator = Validator {
            etag: response.headers().get(ETAG).and_then(|v| v.to_str().ok()).map(str::to_string),
            last_modified: response.headers().get(LAST_MODIFIED).and_then(|v| v.to_str().ok()).map(str::to_string),
        };
        parts.known.lock().unwrap().insert(part.to_path_buf(), validator);
        let mut file = if offset > 0 {
            tokio::fs::OpenOptions::new().append(true).open(part).await
        } else {
            tokio::fs::OpenOptions::new().write(true).create(true).truncate(true).open(part).await
        }
        .map_err(|e| Error::from_io(&e))?;
        return write_body(response, &mut file, offset, total, cancel, progress).await;
    }
    Err(Error::network("服务端的 Range 响应不一致"))
}

async fn write_body(mut response: reqwest::Response, file: &mut tokio::fs::File, offset: u64, total: Option<u64>, cancel: &Cancel, progress: &(dyn Fn(u64, Option<u64>) + Send + Sync)) -> Result<u64> {
    let mut loaded = offset;
    let mut reported = offset;
    progress(loaded, total);
    loop {
        let chunk = tokio::select! {
            c = response.chunk() => c,
            _ = cancel.cancelled() => return Err(Error::aborted()),
        };
        let chunk = match chunk {
            Ok(Some(chunk)) => chunk,
            Ok(None) => break,
            Err(err) => {
                let _ = file.flush().await;
                return Err(net_error(err));
            }
        };
        file.write_all(&chunk).await.map_err(|e| Error::from_io(&e))?;
        loaded += chunk.len() as u64;
        if loaded - reported >= PROGRESS_STEP {
            reported = loaded;
            progress(loaded, total);
        }
    }
    file.flush().await.map_err(|e| Error::from_io(&e))?;
    file.sync_data().await.map_err(|e| Error::from_io(&e))?;
    if total.is_some_and(|t| t != loaded) {
        return Err(Error::network("下载不完整"));
    }
    if reported != loaded {
        progress(loaded, total);
    }
    Ok(loaded)
}

/// `bytes 100-199/1000` → (100, Some(1000))；总长为 * 时为 None。
fn parse_content_range(value: &str) -> Option<(u64, Option<u64>)> {
    let rest = value.trim().strip_prefix("bytes ")?;
    let (range, total) = rest.split_once('/')?;
    let (start, _) = range.split_once('-')?;
    Some((start.trim().parse().ok()?, total.trim().parse().ok()))
}

/// `bytes */1000` → 1000。
fn parse_unsatisfied(value: &str) -> Option<u64> {
    value.trim().strip_prefix("bytes */")?.trim().parse().ok()
}

/// 保存用的文件名：去掉路径（只取最后一段）、控制字符、改变文字方向的字符和各系统不允许的字符，
/// 去掉首尾的空白和点（Windows 会悄悄去掉结尾的点和空格；开头的点在 macOS、Linux 上是隐藏文件），
/// Windows 的保留名（CON、NUL、COM1……，带不带扩展名都算）前加下划线，过长的保留扩展名截断。
pub fn sanitize_file_name(name: &str, fallback: &str) -> String {
    let base = name.rsplit(['/', '\\']).next().unwrap_or("");
    let cleaned: String = base
        .chars()
        .filter(|c| !c.is_control() && !matches!(*c, '\u{202a}'..='\u{202e}' | '\u{2066}'..='\u{2069}' | '\u{200e}' | '\u{200f}'))
        .map(|c| if matches!(c, '<' | '>' | ':' | '"' | '|' | '?' | '*') { '_' } else { c })
        .collect();
    let trimmed = cleaned.trim_matches(|c: char| c.is_whitespace() || c == '.');
    let mut out = if trimmed.is_empty() { fallback.to_string() } else { trimmed.to_string() };
    if is_reserved(&out) {
        out.insert(0, '_');
    }
    truncate_keep_ext(out, MAX_NAME_BYTES)
}

fn is_reserved(name: &str) -> bool {
    let stem = name.split('.').next().unwrap_or("").trim_end().to_ascii_uppercase();
    matches!(stem.as_str(), "CON" | "PRN" | "AUX" | "NUL" | "CONIN$" | "CONOUT$")
        || ((stem.starts_with("COM") || stem.starts_with("LPT")) && {
            let tail = &stem[3..];
            matches!(tail, "0" | "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8" | "9" | "¹" | "²" | "³")
        })
}

fn split_ext(name: &str) -> (&str, &str) {
    match name.rfind('.') {
        Some(i) if i > 0 && name.len() - i <= 17 => (&name[..i], &name[i..]),
        _ => (name, ""),
    }
}

fn truncate_keep_ext(name: String, max: usize) -> String {
    if name.len() <= max {
        return name;
    }
    let (stem, ext) = split_ext(&name);
    let mut end = max.saturating_sub(ext.len()).min(stem.len());
    while !stem.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}{}", stem[..end].trim_end_matches([' ', '.']), ext)
}

/// 重名时的第 n 个名称：“名称 (n).扩展名”。
pub fn numbered(name: &str, n: u32) -> String {
    let (stem, ext) = split_ext(name);
    format!("{stem} ({n}){ext}")
}

/// target 已被占用（文件存在，或另一个下载正在写）时，依次试“名称 (1).扩展名”……
pub fn next_free(target: &Path, busy: impl Fn(&Path) -> bool) -> PathBuf {
    let name = target.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();
    for n in 1..10_000 {
        let candidate = target.with_file_name(numbered(&name, n));
        if !exists(&candidate) && !busy(&candidate) {
            return candidate;
        }
    }
    target.with_file_name(format!("{} {name}", super::read::random_hex()))
}

/// 下载的目标：conflict 为 rename 时避开已有的文件和正在下载的目标；overwrite 时只避开正在下载的目标。
pub fn choose_target(dir: &Path, name: &str, rename: bool, busy: impl Fn(&Path) -> bool) -> PathBuf {
    let first = dir.join(name);
    if busy(&first) || (rename && exists(&first)) {
        return next_free(&first, busy);
    }
    first
}

/// 地址中的文件名（没有 name 参数时）：路径的最后一段，百分号解码。
pub fn name_from_url(url: &str) -> Option<String> {
    let path = url.split(['?', '#']).next()?;
    let path = path.split_once("://").map(|(_, rest)| rest).unwrap_or(path);
    let last = path.split_once('/').map(|(_, p)| p)?.rsplit('/').next()?;
    let decoded = percent_encoding::percent_decode_str(last).decode_utf8_lossy().into_owned();
    (!decoded.trim().is_empty()).then_some(decoded)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn file_names_are_sanitized() {
        assert_eq!(sanitize_file_name("..\\..\\evil.exe", "download"), "evil.exe");
        assert_eq!(sanitize_file_name("../../etc/passwd", "download"), "passwd");
        assert_eq!(sanitize_file_name("..", "download"), "download");
        assert_eq!(sanitize_file_name("", "download"), "download");
        assert_eq!(sanitize_file_name("CON.txt", "d"), "_CON.txt");
        assert_eq!(sanitize_file_name("con", "d"), "_con");
        assert_eq!(sanitize_file_name("nul.tar.gz", "d"), "_nul.tar.gz");
        assert_eq!(sanitize_file_name("COM1.log", "d"), "_COM1.log");
        assert_eq!(sanitize_file_name("lpt9", "d"), "_lpt9");
        assert_eq!(sanitize_file_name("COM10.txt", "d"), "COM10.txt");
        assert_eq!(sanitize_file_name("console.txt", "d"), "console.txt");
        assert_eq!(sanitize_file_name("a\u{0}b\u{1f}c\u{7f}.txt", "d"), "abc.txt");
        assert_eq!(sanitize_file_name("invoice\u{202e}fdp.exe", "d"), "invoicefdp.exe");
        assert_eq!(sanitize_file_name("a<b>c:d\"e|f?g*.txt", "d"), "a_b_c_d_e_f_g_.txt");
        assert_eq!(sanitize_file_name("report.pdf. . ", "d"), "report.pdf");
        assert_eq!(sanitize_file_name(".bashrc", "d"), "bashrc");
        assert_eq!(sanitize_file_name("  报告 .docx ", "d"), "报告 .docx");
        let long = format!("{}.pdf", "文".repeat(200));
        let cut = sanitize_file_name(&long, "d");
        assert!(cut.len() <= MAX_NAME_BYTES && cut.ends_with(".pdf") && cut.starts_with('文'));
        let long_ext = format!("{}.{}", "a".repeat(300), "b".repeat(40));
        assert!(sanitize_file_name(&long_ext, "d").len() <= MAX_NAME_BYTES);
    }

    #[test]
    fn conflict_names() {
        assert_eq!(numbered("名称.pdf", 1), "名称 (1).pdf");
        assert_eq!(numbered("archive.tar.gz", 2), "archive.tar (2).gz");
        assert_eq!(numbered("README", 3), "README (3)");
        let dir = tempfile::tempdir().unwrap();
        let d = dir.path();
        assert_eq!(choose_target(d, "a.txt", true, |_| false), d.join("a.txt"));
        std::fs::write(d.join("a.txt"), b"1").unwrap();
        std::fs::write(d.join("a (1).txt"), b"2").unwrap();
        assert_eq!(choose_target(d, "a.txt", true, |_| false), d.join("a (2).txt"));
        assert_eq!(choose_target(d, "a.txt", false, |_| false), d.join("a.txt"));
        // 正在下载的目标同样避开
        let busy = d.join("a (2).txt");
        assert_eq!(choose_target(d, "a.txt", true, |p| p == busy), d.join("a (3).txt"));
        let first = d.join("b.txt");
        assert_eq!(choose_target(d, "b.txt", false, |p| p == first), d.join("b (1).txt"));
        // 只有 .part（本次下载中断留下的）不算重名
        std::fs::write(d.join("c.bin.part"), b"x").unwrap();
        assert_eq!(choose_target(d, "c.bin", true, |_| false), d.join("c.bin"));
        assert_eq!(part_path(&d.join("c.bin")), d.join("c.bin.part"));
    }

    #[test]
    fn url_names_and_ranges() {
        assert_eq!(name_from_url("https://s3.example.com/b/%E6%8A%A5%E5%91%8A.pdf?X-Amz-Signature=abc").as_deref(), Some("报告.pdf"));
        assert_eq!(name_from_url("https://example.com/").as_deref(), None);
        assert_eq!(name_from_url("https://example.com").as_deref(), None);
        assert_eq!(parse_content_range("bytes 100-199/1000"), Some((100, Some(1000))));
        assert_eq!(parse_content_range("bytes 0-9/*"), Some((0, None)));
        assert_eq!(parse_unsatisfied("bytes */1000"), Some(1000));
        let mut buf = *b"0123456789";
        apply_patches(&mut buf, 100, &[98, 104, 109]);
        assert_eq!(&buf, b"ee23free8f");
    }
}