libfw-client 0.1.3

//! Download scheduler: recursive folder listing, per-file streaming
//! downloads with Range/If-Range resume, exponential-backoff retries and
//! bounded-memory zrip decompression.

use std::cell::{Cell, RefCell};
use std::rc::Rc;

use js_sys::Reflect;
use wasm_bindgen::JsCast;
use wasm_bindgen::JsValue;
use web_sys::Response;

use futures::StreamExt;
use libfw_core::compress::{decompressor, CompressionFormat};
use libfw_core::storage::DirEntry;
use libfw_core::HEADER_COMPRESS;

use crate::config::ClientConfig;
use crate::error::LibfwError;
use crate::http::{auth_headers, dir_url, fetch, file_url, read_all, request, stream_body};
use crate::js::Callbacks;
use crate::plan::{total_bytes, FileEntry};
use crate::state::TaskControl;

/// A single file download outcome, used for resume-state bookkeeping.
struct DownloadOutcome {
    /// Final absolute size/offset observed from the server.
    size: u64,
}

/// Persist download progress roughly every this many bytes so an
/// interrupted transfer can resume from a recent offset instead of
/// restarting from byte 0.
const RESUME_SAVE_EVERY: u64 = 4 * 1024 * 1024;

/// List the immediate children of a virtual directory via `GET /dir/..`.
async fn list_dir(
    base_url: &str,
    token: &str,
    path: &str,
    timeout_ms: u32,
) -> Result<Vec<DirEntry>, LibfwError> {
    let headers = auth_headers(token, false)?;
    let url = dir_url(base_url, path);
    let req = request(&url, "GET", &headers, None)?;
    let resp = fetch(&req, timeout_ms).await?;
    let status = resp.status();
    if status != 200 {
        return Err(LibfwError::Http { status, url });
    }
    let body = read_all(&resp, timeout_ms).await?;
    serde_json::from_slice::<Vec<DirEntry>>(&body)
        .map_err(|e| LibfwError::Protocol(format!("bad listing JSON: {e}")))
}

/// Recursively collect every file under `path` (server-side walk).
///
/// Iterative with an explicit stack so deep directory trees neither blow
/// the compiler recursion limit nor the WASM call stack.
async fn collect_files(
    base_url: &str,
    token: &str,
    path: &str,
    timeout_ms: u32,
) -> Result<Vec<FileEntry>, LibfwError> {
    let mut out = Vec::new();
    let mut stack = vec![path.to_string()];
    while let Some(dir) = stack.pop() {
        for entry in list_dir(base_url, token, &dir, timeout_ms).await? {
            if entry.is_dir {
                stack.push(entry.path);
            } else {
                out.push(FileEntry {
                    path: entry.path,
                    size: entry.size,
                    mtime: entry.mtime,
                });
            }
        }
    }
    out.sort_by(|a, b| a.path.cmp(&b.path));
    Ok(out)
}

/// Sleep for `ms` milliseconds on the JS event loop.
async fn sleep_ms(ms: u32) {
    if ms == 0 {
        return;
    }
    let promise = js_sys::Promise::new(&mut |resolve, _reject| {
        let window = web_sys::window().expect("window");
        let f: &js_sys::Function = resolve.unchecked_ref();
        let _ = window.set_timeout_with_callback_and_timeout_and_arguments_0(f, ms as i32);
    });
    let _ = wasm_bindgen_futures::JsFuture::from(promise).await;
}

/// Download one file with resume + retry.
async fn download_file(
    base_url: &str,
    token: &str,
    file: &FileEntry,
    callbacks: &Callbacks,
    control: &TaskControl,
    config: &ClientConfig,
) -> Result<DownloadOutcome, LibfwError> {
    callbacks.on_file_start(&file.path, file.size)?;

    // 1. Load persisted resume state: { etag, offset }.
    let mut resume: Option<(String, u64)> = None;
    if let Some(state) = callbacks.load_state("download", &file.path).await? {
        let etag = Reflect::get(&state, &JsValue::from_str("etag"))
            .ok()
            .and_then(|v| v.as_string())
            .unwrap_or_default();
        let offset = Reflect::get(&state, &JsValue::from_str("offset"))
            .ok()
            .and_then(|v| v.as_f64())
            .unwrap_or(0.0) as u64;
        if !etag.is_empty() && offset > 0 {
            resume = Some((etag, offset));
        }
    }

    let mut offset = resume.as_ref().map(|(_, o)| *o).unwrap_or(0);
    let mut etag = resume.as_ref().map(|(e, _)| e.clone()).unwrap_or_default();
    let mut attempts = 0u32;

    loop {
        control.wait_ready().await?;
        control.check()?;

        // Re-validate the offset on every (re)try via If-Range.
        let headers = auth_headers(token, config.compress)?;
        headers
            .set("Range", &format!("bytes={offset}-"))
            .map_err(|e| LibfwError::Js(format!("set Range failed: {e:?}")))?;
        if offset > 0 && !etag.is_empty() {
            headers
                .set("If-Range", &etag)
                .map_err(|e| LibfwError::Js(format!("set If-Range failed: {e:?}")))?;
        }
        let url = file_url(base_url, &file.path);
        let req = request(&url, "GET", &headers, None)?;

        match fetch(&req, config.timeout_ms).await {
            Ok(resp) => match resp.status() {
                200 => {
                    // Full content: the file changed (or first attempt).
                    etag = response_etag(&resp).unwrap_or(etag);
                    let outcome =
                        stream_download(&resp, file, callbacks, control, 0, &etag, config.timeout_ms)
                            .await?;
                    return finish_download(file, callbacks, etag, outcome).await;
                }
                206 => {
                    if etag.is_empty() {
                        etag = response_etag(&resp).unwrap_or_default();
                    }
                    let start = content_range_start(&resp).unwrap_or(offset);
                    let outcome = stream_download(
                        &resp,
                        file,
                        callbacks,
                        control,
                        start,
                        &etag,
                        config.timeout_ms,
                    )
                    .await?;
                    return finish_download(file, callbacks, etag, outcome).await;
                }
                416 => {
                    // Offset beyond EOF → the file shrank; restart cleanly.
                    offset = 0;
                    attempts = 0;
                    continue;
                }
                code => return Err(LibfwError::Http { status: code, url }),
            },
            Err(e) => {
                // Network failure → exponential backoff and retry.
                if attempts >= config.max_retries {
                    return Err(e);
                }
                attempts += 1;
                callbacks.log(&format!(
                    "retrying `{}` (attempt {attempts}): {e}",
                    file.path
                ));
                sleep_ms(config.backoff_ms(attempts)).await;
            }
        }
    }
}

/// Stream a `200`/`206` response body, decompressing on the fly and
/// pushing chunks to JS. `start` is the byte offset the body begins at.
///
/// Progress is persisted to the resume store every [`RESUME_SAVE_EVERY`]
/// bytes (best-effort) so an interrupted transfer can resume from a recent
/// offset; `finish_download` persists the final state on success.
async fn stream_download(
    resp: &Response,
    file: &FileEntry,
    callbacks: &Callbacks,
    control: &TaskControl,
    start: u64,
    etag: &str,
    timeout_ms: u32,
) -> Result<DownloadOutcome, LibfwError> {
    // Decide the wire format from the response header (robust against a
    // server that did not honour our Accept-Encoding).
    let format = resp
        .headers()
        .get(HEADER_COMPRESS)
        .ok()
        .flatten()
        .and_then(|v| CompressionFormat::parse_header(&v))
        .unwrap_or(CompressionFormat::None);

    // State is shared via `Rc` so the per-chunk `FnMut` callback can move
    // owned clones into its `async move` block (single-threaded WASM).
    let decomp = Rc::new(RefCell::new(decompressor(format)));
    let out = Rc::new(RefCell::new(Vec::new()));
    let file_offset = Rc::new(Cell::new(start));
    let last_saved = Rc::new(Cell::new(0u64));
    let callbacks = callbacks.clone();
    let control = control.clone();
    let path = file.path.clone();
    let etag = etag.to_string();
    let final_size = file.size;

    stream_body(
        resp,
        timeout_ms,
        |chunk| {
            let decomp = decomp.clone();
            let out = out.clone();
            let file_offset = file_offset.clone();
            let last_saved = last_saved.clone();
            let callbacks = callbacks.clone();
            let control = control.clone();
            let path = path.clone();
            let etag = etag.clone();
            async move {
                control.wait_ready().await?;
                control.check()?;
                decomp
                    .borrow_mut()
                    .decompress(&chunk, &mut out.borrow_mut())
                    .map_err(|e| LibfwError::Decompress(e.to_string()))?;
                let data = std::mem::take(&mut *out.borrow_mut());
                if !data.is_empty() {
                    let offset = file_offset.get();
                    callbacks.on_write_chunk(&path, offset, &data).await?;
                    file_offset.set(offset.saturating_add(data.len() as u64));
                    control.add_progress(data.len() as u64);

                    // Persist an absolute resume offset every so often so a
                    // crash mid-transfer can continue instead of restarting.
                    let done = file_offset.get();
                    if done >= last_saved.get().saturating_add(RESUME_SAVE_EVERY) {
                        last_saved.set(done);
                        let _ = callbacks
                            .save_state("download", &path, &resume_state_obj(&etag, done))
                            .await;
                    }
                }
                Ok(())
            }
        },
    )
    .await?;

    // Flush any final decompressed frames.
    decomp
        .borrow_mut()
        .finish(&mut out.borrow_mut())
        .map_err(|e| LibfwError::Decompress(e.to_string()))?;
    let tail = std::mem::take(&mut *out.borrow_mut());
    if !tail.is_empty() {
        let offset = file_offset.get();
        callbacks.on_write_chunk(&path, offset, &tail).await?;
        file_offset.set(offset.saturating_add(tail.len() as u64));
        control.add_progress(tail.len() as u64);
    }

    Ok(DownloadOutcome {
        size: final_size.max(file_offset.get()),
    })
}

/// Build a resume-state object for JS persistence.
fn resume_state_obj(etag: &str, offset: u64) -> JsValue {
    let state = js_sys::Object::new();
    let _ = js_sys::Reflect::set(
        &state,
        &JsValue::from_str("etag"),
        &JsValue::from_str(etag),
    );
    let _ = js_sys::Reflect::set(
        &state,
        &JsValue::from_str("offset"),
        &JsValue::from_f64(offset as f64),
    );
    let _ = js_sys::Reflect::set(
        &state,
        &JsValue::from_str("size"),
        &JsValue::from_f64(offset as f64),
    );
    state.into()
}

/// Persist resume state (via the JS IndexedDB layer) and notify JS.
async fn finish_download(
    file: &FileEntry,
    callbacks: &Callbacks,
    etag: String,
    outcome: DownloadOutcome,
) -> Result<DownloadOutcome, LibfwError> {
    // Persist the ABSOLUTE end offset (`size`), not the per-request delta,
    // so a later resume re-requests the correct byte range.
    let offset = outcome.size;
    let size = outcome.size;
    let state = js_sys::Object::new();
    js_sys::Reflect::set(&state, &JsValue::from_str("etag"), &JsValue::from_str(&etag))
        .map_err(|e| LibfwError::Js(format!("state etag: {e:?}")))?;
    js_sys::Reflect::set(&state, &JsValue::from_str("offset"), &JsValue::from_f64(offset as f64))
        .map_err(|e| LibfwError::Js(format!("state offset: {e:?}")))?;
    js_sys::Reflect::set(&state, &JsValue::from_str("size"), &JsValue::from_f64(size as f64))
        .map_err(|e| LibfwError::Js(format!("state size: {e:?}")))?;
    callbacks.save_state("download", &file.path, &state).await?;
    callbacks.on_file_completed(&file.path).await?;
    Ok(outcome)
}

/// Download an entire folder (or the root when `path` is empty).
pub async fn download_folder(
    base_url: &str,
    token: &str,
    path: &str,
    callbacks: &Callbacks,
    control: &TaskControl,
    config: &ClientConfig,
) -> Result<u64, LibfwError> {
    let files = collect_files(base_url, token, path, config.timeout_ms).await?;
    let total = total_bytes(&files);
    control.set_total(total);
    callbacks.on_progress(0, total)?;

    let mut stream = futures::stream::iter(files.into_iter().map(|file| {
        let base_url = base_url.to_string();
        let token = token.to_string();
        let callbacks = callbacks.clone();
        let control = control.clone();
        let config = config.clone();
        async move { download_file(&base_url, &token, &file, &callbacks, &control, &config).await }
    }))
    .buffer_unordered(config.concurrency);

    while let Some(result) = stream.next().await {
        result?;
        // Report progress from the shared control block so pause/resume and
        // the onProgress events stay consistent (one source of truth).
        callbacks.on_progress(control.done_bytes(), control.total_bytes())?;
    }
    Ok(control.done_bytes())
}

/// Download a single file at `path` (size/etag discovered from the server).
pub async fn download_single(
    base_url: &str,
    token: &str,
    path: &str,
    callbacks: &Callbacks,
    control: &TaskControl,
    config: &ClientConfig,
) -> Result<u64, LibfwError> {
    let file = FileEntry {
        path: path.to_string(),
        size: 0,
        mtime: 0,
    };
    let outcome = download_file(base_url, token, &file, callbacks, control, config).await?;
    // Return the ABSOLUTE byte count (the final offset), consistent with
    // `download_folder`, rather than this response's delta (which would be
    // misleading on a resumed download).
    Ok(outcome.size)
}

/// Read the ETag response header.
fn response_etag(resp: &Response) -> Option<String> {
    resp.headers().get("etag").ok().flatten()
}

/// Parse `Content-Range: bytes start-end/total` → `start`.
fn content_range_start(resp: &Response) -> Option<u64> {
    let value = resp.headers().get("content-range").ok().flatten()?;
    let after = value.split_whitespace().nth(1)?; // "start-end/total"
    let start = after.split('-').next()?;
    start.parse().ok()
}