Skip to main content

kget/
builder.rs

1//! Fluent builder API — the primary interface for the KGet library.
2//!
3//! # Quick start
4//!
5//! ```rust,no_run
6//! use kget::KgetError;
7//!
8//! // Single file
9//! let result = kget::builder("https://example.com/file.zip")
10//!     .output("./downloads/")
11//!     .connections(8)
12//!     .sha256("abc123...")
13//!     .download()?;
14//!
15//! println!("Avg speed: {} B/s", result.avg_speed_bps);
16//! # Ok::<(), KgetError>(())
17//! ```
18//!
19//! # Batch
20//!
21//! ```rust,no_run
22//! # use kget::KgetError;
23//! let results = kget::batch(["https://a.com/f1.zip", "https://b.com/f2.iso"])
24//!     .concurrency(4)
25//!     .output_dir("./downloads/")
26//!     .download_all();
27//!
28//! for r in &results {
29//!     match &r.result {
30//!         Ok(info) => println!("OK  {}", r.url),
31//!         Err(e)   => eprintln!("ERR {}  — {}", r.url, e),
32//!     }
33//! }
34//! ```
35
36use crate::DownloadOptions;
37use crate::advanced_download::AdvancedDownloader;
38use crate::checksum::{ChecksumAlgorithm, compute_checksum, parse_sidecar};
39use crate::config::{Config, ProxyConfig, ProxyType};
40use crate::download::download as http_download;
41use crate::error::KgetError;
42use crate::events::DownloadEvent;
43use crate::optimization::Optimizer;
44use crate::utils;
45use std::io::{Cursor, Read};
46use std::path::Path;
47use std::sync::Arc;
48use std::sync::mpsc;
49use std::thread;
50use std::time::{Duration, Instant};
51
52// ════════════════════════════════════════════════════════════════════════════
53// Retry configuration
54// ════════════════════════════════════════════════════════════════════════════
55
56/// Delay strategy between retry attempts.
57#[derive(Debug, Clone)]
58pub enum Backoff {
59    /// Always wait the same fixed duration.
60    Fixed(Duration),
61    /// Double the delay each attempt, capped at `max_ms`.
62    Exponential {
63        /// Initial delay in milliseconds.
64        base_ms: u64,
65        /// Maximum delay cap in milliseconds.
66        max_ms: u64,
67    },
68}
69
70impl Backoff {
71    fn delay(&self, attempt: u32) -> Duration {
72        match self {
73            Backoff::Fixed(d) => *d,
74            Backoff::Exponential { base_ms, max_ms } => {
75                let ms = base_ms.saturating_mul(1u64 << attempt.min(62));
76                Duration::from_millis(ms.min(*max_ms))
77            }
78        }
79    }
80}
81
82/// Controls how many times and how often a failed download is retried.
83#[derive(Debug, Clone)]
84pub struct RetryConfig {
85    /// Maximum total attempts (including the first).  Default: 3.
86    pub max_attempts: u32,
87    /// Delay strategy between retries.
88    pub backoff: Backoff,
89    /// HTTP status codes that should trigger a retry.
90    /// Non-HTTP errors (IO, cancel) always abort immediately.
91    pub retry_on_status: Vec<u16>,
92}
93
94impl Default for RetryConfig {
95    fn default() -> Self {
96        RetryConfig {
97            max_attempts: 3,
98            backoff: Backoff::Exponential { base_ms: 500, max_ms: 30_000 },
99            retry_on_status: vec![408, 429, 500, 502, 503, 504],
100        }
101    }
102}
103
104// ════════════════════════════════════════════════════════════════════════════
105// Download result
106// ════════════════════════════════════════════════════════════════════════════
107
108/// Digests that were computed (and optionally verified) after a download.
109#[derive(Debug, Clone, Default)]
110pub struct ComputedChecksums {
111    pub sha256: Option<String>,
112    pub sha512: Option<String>,
113    pub sha1: Option<String>,
114    pub md5: Option<String>,
115    pub blake3: Option<String>,
116}
117
118/// Metrics and metadata returned after a successful download.
119#[derive(Debug, Clone)]
120pub struct DownloadResult {
121    /// Absolute path of the saved file.  Empty for in-memory downloads.
122    pub path: String,
123    /// Total bytes written to disk.
124    pub bytes_downloaded: u64,
125    /// Average transfer speed in bytes per second.
126    pub avg_speed_bps: u64,
127    /// Wall-clock time from start to completion.
128    pub duration: Duration,
129    /// Number of parallel connections used.
130    pub connections_used: usize,
131    /// Checksums that were computed during or after the download.
132    pub checksums: ComputedChecksums,
133}
134
135// ════════════════════════════════════════════════════════════════════════════
136// DownloadBuilder internals
137// ════════════════════════════════════════════════════════════════════════════
138
139#[derive(Debug, Clone, Default)]
140struct ChecksumExpectations {
141    sha256: Option<String>,
142    sha512: Option<String>,
143    sha1: Option<String>,
144    md5: Option<String>,
145    blake3: Option<String>,
146}
147
148impl ChecksumExpectations {
149    fn any_set(&self) -> bool {
150        self.sha256.is_some()
151            || self.sha512.is_some()
152            || self.sha1.is_some()
153            || self.md5.is_some()
154            || self.blake3.is_some()
155    }
156}
157
158// ════════════════════════════════════════════════════════════════════════════
159// DownloadBuilder
160// ════════════════════════════════════════════════════════════════════════════
161
162/// Fluent builder for a single download.
163///
164/// Create one via [`crate::builder()`] or [`DownloadBuilder::new()`].
165pub struct DownloadBuilder {
166    url: String,
167    output: Option<String>,
168    connections: usize,
169    speed_limit: Option<u64>,
170    proxy_url: Option<String>,
171    proxy_user: Option<String>,
172    proxy_pass: Option<String>,
173    checksums: ChecksumExpectations,
174    verify_from: Option<String>,
175    headers: Vec<(String, String)>,
176    retry: RetryConfig,
177    range: Option<(u64, u64)>,
178    quiet: bool,
179}
180
181impl DownloadBuilder {
182    /// Start building a download for `url`.
183    pub fn new(url: impl Into<String>) -> Self {
184        DownloadBuilder {
185            url: url.into(),
186            output: None,
187            connections: 1,
188            speed_limit: None,
189            proxy_url: None,
190            proxy_user: None,
191            proxy_pass: None,
192            checksums: ChecksumExpectations::default(),
193            verify_from: None,
194            headers: Vec::new(),
195            retry: RetryConfig::default(),
196            range: None,
197            quiet: false,
198        }
199    }
200
201    // ── Configuration ────────────────────────────────────────────────────────
202
203    /// Set the output file path or directory.
204    ///
205    /// If a directory is given, the filename is inferred from the URL.
206    pub fn output(mut self, path: impl Into<String>) -> Self {
207        self.output = Some(path.into());
208        self
209    }
210
211    /// Number of parallel HTTP connections.  Clamped to 1–32.
212    pub fn connections(mut self, n: usize) -> Self {
213        self.connections = n.clamp(1, 32);
214        self
215    }
216
217    /// Global speed limit in bytes per second.
218    pub fn speed_limit(mut self, bytes_per_sec: u64) -> Self {
219        self.speed_limit = Some(bytes_per_sec);
220        self
221    }
222
223    /// HTTP proxy URL (e.g. `"http://proxy:8080"` or `"socks5://host:1080"`).
224    pub fn proxy(mut self, url: impl Into<String>) -> Self {
225        self.proxy_url = Some(url.into());
226        self
227    }
228
229    /// Proxy credentials (optional, required only for authenticated proxies).
230    pub fn proxy_auth(mut self, user: impl Into<String>, pass: impl Into<String>) -> Self {
231        self.proxy_user = Some(user.into());
232        self.proxy_pass = Some(pass.into());
233        self
234    }
235
236    /// Expect a SHA-256 digest.  The download fails if the file doesn't match.
237    pub fn sha256(mut self, hash: impl Into<String>) -> Self {
238        self.checksums.sha256 = Some(hash.into().to_lowercase());
239        self
240    }
241
242    /// Expect a SHA-512 digest.
243    pub fn sha512(mut self, hash: impl Into<String>) -> Self {
244        self.checksums.sha512 = Some(hash.into().to_lowercase());
245        self
246    }
247
248    /// Expect a SHA-1 digest.
249    pub fn sha1(mut self, hash: impl Into<String>) -> Self {
250        self.checksums.sha1 = Some(hash.into().to_lowercase());
251        self
252    }
253
254    /// Expect an MD5 digest.
255    pub fn md5(mut self, hash: impl Into<String>) -> Self {
256        self.checksums.md5 = Some(hash.into().to_lowercase());
257        self
258    }
259
260    /// Expect a BLAKE3 digest.
261    pub fn blake3(mut self, hash: impl Into<String>) -> Self {
262        self.checksums.blake3 = Some(hash.into().to_lowercase());
263        self
264    }
265
266    /// Download a checksum sidecar file from `url` and use it to populate the
267    /// expected digest automatically.
268    ///
269    /// Supports GNU (`<hash>  <file>`) and BSD (`SHA256 (<file>) = <hash>`) formats.
270    pub fn verify_from(mut self, url: impl Into<String>) -> Self {
271        self.verify_from = Some(url.into());
272        self
273    }
274
275    /// Add a custom HTTP request header.  Can be called multiple times.
276    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
277        self.headers.push((name.into(), value.into()));
278        self
279    }
280
281    /// Override the default retry policy.
282    pub fn retry(mut self, config: RetryConfig) -> Self {
283        self.retry = config;
284        self
285    }
286
287    /// Request only a byte range of the file: `[start, end]` (both inclusive).
288    ///
289    /// Sends `Range: bytes=start-end`.  The returned `DownloadResult` will have
290    /// `bytes_downloaded` equal to `end - start + 1`.
291    pub fn range(mut self, start: u64, end: u64) -> Self {
292        self.range = Some((start, end));
293        self
294    }
295
296    /// Suppress all progress output.
297    pub fn quiet(mut self, q: bool) -> Self {
298        self.quiet = q;
299        self
300    }
301
302    // ── Terminal methods ─────────────────────────────────────────────────────
303
304    /// Execute the download synchronously and return metrics on success.
305    pub fn download(mut self) -> Result<DownloadResult, KgetError> {
306        // 1. Resolve sidecar before the main download so the hash is ready.
307        if let Some(sidecar_url) = self.verify_from.take() {
308            self.apply_sidecar(&sidecar_url)?;
309        }
310
311        let output_path = self.resolve_output();
312        let proxy = self.make_proxy();
313        let optimizer = self.make_optimizer();
314        let start = Instant::now();
315
316        // 2. Execute the download (with retry).
317        self.run_with_retry(&output_path, proxy.clone(), optimizer.clone())?;
318
319        let duration = start.elapsed();
320
321        // 3. Verify checksums and collect digests.
322        let checksums = self.verify_and_collect(Path::new(&output_path))?;
323
324        // 4. Build result metrics.
325        let bytes_downloaded = std::fs::metadata(&output_path)
326            .map(|m| m.len())
327            .unwrap_or(0);
328        let avg_speed_bps = if duration.as_secs() > 0 {
329            bytes_downloaded / duration.as_secs()
330        } else {
331            bytes_downloaded
332        };
333
334        Ok(DownloadResult {
335            path: output_path,
336            bytes_downloaded,
337            avg_speed_bps,
338            duration,
339            connections_used: self.connections,
340            checksums,
341        })
342    }
343
344    /// Download entirely into memory and return the raw bytes.
345    ///
346    /// Respects `.range()`, `.proxy()`, and `.header()` settings.
347    /// Does **not** write to disk, so `.output()` is ignored.
348    pub fn download_to_bytes(self) -> Result<Vec<u8>, KgetError> {
349        let client = self.make_blocking_client()?;
350        let mut req = client.get(&self.url);
351
352        if let Some((s, e)) = self.range {
353            req = req.header("Range", format!("bytes={s}-{e}"));
354        }
355        req = apply_headers(req, &self.headers);
356
357        let resp = req.send()?;
358        if !resp.status().is_success() && resp.status().as_u16() != 206 {
359            if resp.status().as_u16() == 404 {
360                return Err(KgetError::NotFound(self.url.clone()));
361            }
362            return Err(KgetError::Network(format!(
363                "HTTP {} for {}",
364                resp.status(),
365                self.url
366            )));
367        }
368        Ok(resp.bytes()?.to_vec())
369    }
370
371    /// Download to memory and return an `impl Read`.
372    ///
373    /// Convenience wrapper over [`download_to_bytes`](Self::download_to_bytes).
374    pub fn download_to_reader(self) -> Result<impl Read, KgetError> {
375        self.download_to_bytes().map(Cursor::new)
376    }
377
378    /// Spawn the download in a background thread and return an event channel.
379    ///
380    /// The [`DownloadEvent`] receiver yields progress updates until the channel
381    /// is closed (on completion or error).  Call `.join()` on the handle to
382    /// retrieve the final [`DownloadResult`].
383    pub fn spawn(
384        mut self,
385    ) -> (
386        thread::JoinHandle<Result<DownloadResult, KgetError>>,
387        mpsc::Receiver<DownloadEvent>,
388    ) {
389        let (tx, rx) = mpsc::channel::<DownloadEvent>();
390        let handle = thread::spawn(move || {
391            // Resolve sidecar
392            if let Some(sidecar_url) = self.verify_from.take() {
393                if let Err(e) = self.apply_sidecar(&sidecar_url) {
394                    let _ = tx.send(DownloadEvent::Error(e.to_string()));
395                    return Err(e);
396                }
397            }
398
399            let output_path = self.resolve_output();
400            let proxy = self.make_proxy();
401            let optimizer = self.make_optimizer();
402            let start = Instant::now();
403
404            let tx_progress = tx.clone();
405            let tx_status = tx.clone();
406
407            let result =
408                self.run_with_events(&output_path, proxy, optimizer, tx_progress, tx_status);
409
410            match result {
411                Ok(()) => {
412                    let duration = start.elapsed();
413                    let checksums = match self.verify_and_collect(Path::new(&output_path)) {
414                        Ok(c) => c,
415                        Err(e) => {
416                            let _ = tx.send(DownloadEvent::Error(e.to_string()));
417                            return Err(e);
418                        }
419                    };
420                    let bytes_downloaded = std::fs::metadata(&output_path)
421                        .map(|m| m.len())
422                        .unwrap_or(0);
423                    let avg_speed_bps = if duration.as_secs() > 0 {
424                        bytes_downloaded / duration.as_secs()
425                    } else {
426                        bytes_downloaded
427                    };
428                    let _ = tx.send(DownloadEvent::Completed {
429                        path: output_path.clone(),
430                        sha256: checksums.sha256.clone(),
431                    });
432                    Ok(DownloadResult {
433                        path: output_path,
434                        bytes_downloaded,
435                        avg_speed_bps,
436                        duration,
437                        connections_used: self.connections,
438                        checksums,
439                    })
440                }
441                Err(e) => {
442                    let _ = tx.send(DownloadEvent::Error(e.to_string()));
443                    Err(e)
444                }
445            }
446        });
447        (handle, rx)
448    }
449
450    /// Async version of [`download`](Self::download).
451    ///
452    /// Requires the `async` feature.  Runs the blocking download in a
453    /// `tokio::task::spawn_blocking` thread so it doesn't block the executor.
454    #[cfg(feature = "async")]
455    pub async fn download_async(self) -> Result<DownloadResult, KgetError> {
456        tokio::task::spawn_blocking(move || self.download())
457            .await
458            .map_err(|e| KgetError::Other(e.to_string()))?
459    }
460
461    // ── Private helpers ──────────────────────────────────────────────────────
462
463    /// Fetch the sidecar file and, if a matching hash is found, update
464    /// `self.checksums` so the post-download verification uses it.
465    fn apply_sidecar(&mut self, sidecar_url: &str) -> Result<(), KgetError> {
466        let client = self.make_blocking_client()?;
467        let text = client
468            .get(sidecar_url)
469            .send()
470            .map_err(KgetError::from)?
471            .text()
472            .map_err(KgetError::from)?;
473
474        let filename = utils::get_filename_from_url_or_default(&self.url, "file");
475        match parse_sidecar(&text, &filename) {
476            Some((ChecksumAlgorithm::Sha256, h)) => self.checksums.sha256 = Some(h),
477            Some((ChecksumAlgorithm::Sha512, h)) => self.checksums.sha512 = Some(h),
478            Some((ChecksumAlgorithm::Sha1, h))   => self.checksums.sha1   = Some(h),
479            Some((ChecksumAlgorithm::Md5, h))     => self.checksums.md5    = Some(h),
480            Some((ChecksumAlgorithm::Blake3, h))  => self.checksums.blake3 = Some(h),
481            None => return Err(KgetError::SidecarError(format!(
482                "No entry for '{}' found in sidecar file", filename
483            ))),
484        }
485        Ok(())
486    }
487
488    /// Run the download, retrying on transient failures per `self.retry`.
489    fn run_with_retry(
490        &self,
491        output_path: &str,
492        proxy: ProxyConfig,
493        optimizer: Optimizer,
494    ) -> Result<(), KgetError> {
495        let mut attempt = 0u32;
496        loop {
497            let result = self.run_once(output_path, proxy.clone(), optimizer.clone());
498            match result {
499                Ok(()) => return Ok(()),
500                Err(e) => {
501                    attempt += 1;
502                    if attempt >= self.retry.max_attempts {
503                        return Err(e);
504                    }
505                    // Don't retry unrecoverable errors
506                    if matches!(e, KgetError::Cancelled | KgetError::NotFound(_) | KgetError::ChecksumMismatch { .. }) {
507                        return Err(e);
508                    }
509                    let delay = self.retry.backoff.delay(attempt - 1);
510                    if !self.quiet {
511                        eprintln!(
512                            "Attempt {}/{} failed: {e}. Retrying in {:?}…",
513                            attempt, self.retry.max_attempts, delay
514                        );
515                    }
516                    thread::sleep(delay);
517                }
518            }
519        }
520    }
521
522    /// One attempt at the underlying download.
523    fn run_once(
524        &self,
525        output_path: &str,
526        proxy: ProxyConfig,
527        optimizer: Optimizer,
528    ) -> Result<(), KgetError> {
529        // Range request: bypass the normal downloaders, use reqwest directly.
530        if let Some((range_start, range_end)) = self.range {
531            return self.download_range(output_path, range_start, range_end);
532        }
533
534        if self.connections > 1 {
535            let mut dl = AdvancedDownloader::new(
536                self.url.clone(),
537                output_path.to_string(),
538                self.quiet,
539                proxy,
540                optimizer,
541            )
542            .map_err(KgetError::from)?;
543            dl.set_extra_headers(self.headers.clone());
544            if let Some(h) = &self.checksums.sha256 {
545                dl.set_expected_sha256(h.clone());
546            }
547            dl.download().map_err(KgetError::from)
548        } else {
549            let options = DownloadOptions {
550                quiet_mode: self.quiet,
551                output_path: Some(output_path.to_string()),
552                verify_iso: false,
553                expected_sha256: self.checksums.sha256.clone(),
554                extra_headers: self.headers.clone(),
555            };
556            http_download(&self.url, proxy, optimizer, options, None).map_err(KgetError::from)
557        }
558    }
559
560    /// Like `run_once` but hooks AdvancedDownloader callbacks to the event sender.
561    fn run_with_events(
562        &self,
563        output_path: &str,
564        proxy: ProxyConfig,
565        optimizer: Optimizer,
566        tx_progress: mpsc::Sender<DownloadEvent>,
567        tx_status: mpsc::Sender<DownloadEvent>,
568    ) -> Result<(), KgetError> {
569        if let Some((range_start, range_end)) = self.range {
570            return self.download_range(output_path, range_start, range_end);
571        }
572
573        if self.connections > 1 {
574            let mut dl = AdvancedDownloader::new(
575                self.url.clone(),
576                output_path.to_string(),
577                self.quiet,
578                proxy,
579                optimizer,
580            )
581            .map_err(KgetError::from)?;
582            dl.set_extra_headers(self.headers.clone());
583            if let Some(h) = &self.checksums.sha256 {
584                dl.set_expected_sha256(h.clone());
585            }
586            dl.set_progress_callback(move |p| {
587                let _ = tx_progress.send(DownloadEvent::Progress {
588                    percent: p as f64 * 100.0,
589                    speed_bps: 0,
590                    eta_secs: None,
591                });
592            });
593            dl.set_status_callback(move |msg| {
594                let _ = tx_status.send(DownloadEvent::Status(msg));
595            });
596            dl.download().map_err(KgetError::from)
597        } else {
598            let options = DownloadOptions {
599                quiet_mode: self.quiet,
600                output_path: Some(output_path.to_string()),
601                verify_iso: false,
602                expected_sha256: self.checksums.sha256.clone(),
603                extra_headers: self.headers.clone(),
604            };
605            let status_cb = move |msg: String| {
606                // Parse PROGRESS: lines if present
607                if let Some(pct) = extract_percent(&msg) {
608                    let _ = tx_progress.send(DownloadEvent::Progress {
609                        percent: pct,
610                        speed_bps: 0,
611                        eta_secs: None,
612                    });
613                } else {
614                    let _ = tx_status.send(DownloadEvent::Status(msg));
615                }
616            };
617            http_download(&self.url, proxy, optimizer, options, Some(&status_cb))
618                .map_err(KgetError::from)
619        }
620    }
621
622    /// Raw range download via reqwest, written to `output_path`.
623    fn download_range(&self, output_path: &str, start: u64, end: u64) -> Result<(), KgetError> {
624        let client = self.make_blocking_client()?;
625        let mut req = client
626            .get(&self.url)
627            .header("Range", format!("bytes={start}-{end}"));
628        req = apply_headers(req, &self.headers);
629
630        let resp = req.send()?;
631        if !resp.status().is_success() && resp.status().as_u16() != 206 {
632            return Err(KgetError::Network(format!(
633                "HTTP {} for range request on {}",
634                resp.status(),
635                self.url
636            )));
637        }
638        let bytes = resp.bytes()?;
639
640        // Create parent directory if needed
641        if let Some(parent) = Path::new(output_path).parent() {
642            std::fs::create_dir_all(parent)?;
643        }
644        std::fs::write(output_path, &bytes)?;
645        Ok(())
646    }
647
648    /// Compute all expected checksums and verify them.
649    /// Returns `ComputedChecksums` populated with any digest that was requested.
650    fn verify_and_collect(&self, path: &Path) -> Result<ComputedChecksums, KgetError> {
651        if !self.checksums.any_set() {
652            return Ok(ComputedChecksums::default());
653        }
654
655        let mut computed = ComputedChecksums::default();
656
657        macro_rules! check {
658            ($field:ident, $algo:expr) => {
659                if let Some(expected) = &self.checksums.$field {
660                    let got = compute_checksum(path, &$algo)?;
661                    if got != *expected {
662                        return Err(KgetError::ChecksumMismatch {
663                            algorithm: $algo.name().to_string(),
664                            expected: expected.clone(),
665                            got,
666                        });
667                    }
668                    computed.$field = Some(got);
669                }
670            };
671        }
672
673        check!(sha256, ChecksumAlgorithm::Sha256);
674        check!(sha512, ChecksumAlgorithm::Sha512);
675        check!(sha1,   ChecksumAlgorithm::Sha1);
676        check!(md5,    ChecksumAlgorithm::Md5);
677        check!(blake3, ChecksumAlgorithm::Blake3);
678
679        Ok(computed)
680    }
681
682    fn resolve_output(&self) -> String {
683        utils::resolve_output_path(self.output.clone(), &self.url, "download")
684    }
685
686    fn make_proxy(&self) -> ProxyConfig {
687        match &self.proxy_url {
688            None => ProxyConfig::default(),
689            Some(url) => {
690                let proxy_type = if url.starts_with("socks5://") {
691                    ProxyType::Socks5
692                } else if url.starts_with("https://") {
693                    ProxyType::Https
694                } else {
695                    ProxyType::Http
696                };
697                ProxyConfig {
698                    enabled: true,
699                    url: Some(url.clone()),
700                    username: self.proxy_user.clone(),
701                    password: self.proxy_pass.clone(),
702                    proxy_type,
703                }
704            }
705        }
706    }
707
708    fn make_optimizer(&self) -> Optimizer {
709        let mut cfg = Config::default().optimization;
710        cfg.speed_limit = self.speed_limit;
711        cfg.max_connections = self.connections;
712        Optimizer::from_config(cfg)
713    }
714
715    fn make_blocking_client(&self) -> Result<reqwest::blocking::Client, KgetError> {
716        let mut b = reqwest::blocking::Client::builder()
717            .timeout(Duration::from_secs(60));
718        if let Some(url) = &self.proxy_url {
719            let mut proxy = reqwest::Proxy::all(url.as_str())
720                .map_err(|e| KgetError::Protocol(e.to_string()))?;
721            if let (Some(u), Some(p)) = (&self.proxy_user, &self.proxy_pass) {
722                proxy = proxy.basic_auth(u, p);
723            }
724            b = b.proxy(proxy);
725        }
726        b.build().map_err(|e| KgetError::Protocol(e.to_string()))
727    }
728}
729
730// ════════════════════════════════════════════════════════════════════════════
731// BatchBuilder
732// ════════════════════════════════════════════════════════════════════════════
733
734/// Result of one URL in a batch download.
735pub struct BatchResult {
736    /// The URL that was attempted.
737    pub url: String,
738    /// Success or failure for that URL.
739    pub result: Result<DownloadResult, KgetError>,
740}
741
742/// Builder for downloading multiple URLs concurrently.
743///
744/// Create one via [`crate::batch()`].
745pub struct BatchBuilder {
746    urls: Vec<String>,
747    concurrency: usize,
748    output_dir: String,
749    speed_limit: Option<u64>,
750    proxy_url: Option<String>,
751    proxy_user: Option<String>,
752    proxy_pass: Option<String>,
753    headers: Vec<(String, String)>,
754    retry: RetryConfig,
755    quiet: bool,
756}
757
758impl BatchBuilder {
759    /// Create a batch builder from any iterator of URL strings.
760    pub fn new(urls: impl IntoIterator<Item = impl Into<String>>) -> Self {
761        BatchBuilder {
762            urls: urls.into_iter().map(Into::into).collect(),
763            concurrency: 4,
764            output_dir: ".".to_string(),
765            speed_limit: None,
766            proxy_url: None,
767            proxy_user: None,
768            proxy_pass: None,
769            headers: Vec::new(),
770            retry: RetryConfig::default(),
771            quiet: false,
772        }
773    }
774
775    /// Maximum number of simultaneous downloads.  Default: 4.
776    pub fn concurrency(mut self, n: usize) -> Self {
777        self.concurrency = n.max(1);
778        self
779    }
780
781    /// Directory where all files are saved.  Default: current directory.
782    pub fn output_dir(mut self, dir: impl Into<String>) -> Self {
783        self.output_dir = dir.into();
784        self
785    }
786
787    /// Speed limit applied to each individual download (bytes/s).
788    pub fn speed_limit(mut self, bps: u64) -> Self {
789        self.speed_limit = Some(bps);
790        self
791    }
792
793    /// HTTP proxy URL shared by all downloads.
794    pub fn proxy(mut self, url: impl Into<String>) -> Self {
795        self.proxy_url = Some(url.into());
796        self
797    }
798
799    /// Proxy credentials.
800    pub fn proxy_auth(mut self, user: impl Into<String>, pass: impl Into<String>) -> Self {
801        self.proxy_user = Some(user.into());
802        self.proxy_pass = Some(pass.into());
803        self
804    }
805
806    /// Add a custom HTTP header sent with every download.
807    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
808        self.headers.push((name.into(), value.into()));
809        self
810    }
811
812    /// Retry policy for each individual download.
813    pub fn retry(mut self, config: RetryConfig) -> Self {
814        self.retry = config;
815        self
816    }
817
818    /// Suppress per-download progress output.
819    pub fn quiet(mut self, q: bool) -> Self {
820        self.quiet = q;
821        self
822    }
823
824    /// Run all downloads with bounded concurrency and return one result per URL.
825    ///
826    /// Uses a Rayon thread pool sized to `concurrency`.
827    pub fn download_all(self) -> Vec<BatchResult> {
828        use rayon::prelude::*;
829
830        let pool = rayon::ThreadPoolBuilder::new()
831            .num_threads(self.concurrency)
832            .build()
833            .expect("failed to build rayon thread pool");
834
835        let output_dir = Arc::new(self.output_dir.clone());
836        let proxy_url  = Arc::new(self.proxy_url.clone());
837        let proxy_user = Arc::new(self.proxy_user.clone());
838        let proxy_pass = Arc::new(self.proxy_pass.clone());
839        let headers    = Arc::new(self.headers.clone());
840        let retry      = Arc::new(self.retry.clone());
841        let speed_limit = self.speed_limit;
842        let quiet       = self.quiet;
843
844        pool.install(|| {
845            self.urls
846                .par_iter()
847                .map(|url| {
848                    let filename =
849                        utils::get_filename_from_url_or_default(url, "download");
850                    let output_path = format!(
851                        "{}/{}",
852                        output_dir.trim_end_matches('/'),
853                        filename
854                    );
855
856                    let mut b = DownloadBuilder::new(url)
857                        .output(&output_path)
858                        .quiet(quiet)
859                        .retry((*retry).clone());
860
861                    if let Some(lim) = speed_limit {
862                        b = b.speed_limit(lim);
863                    }
864                    if let Some(pu) = proxy_url.as_ref() {
865                        b = b.proxy(pu.clone());
866                        if let (Some(user), Some(pass)) =
867                            (proxy_user.as_ref(), proxy_pass.as_ref())
868                        {
869                            b = b.proxy_auth(user.clone(), pass.clone());
870                        }
871                    }
872                    for (k, v) in headers.iter() {
873                        b = b.header(k.clone(), v.clone());
874                    }
875
876                    BatchResult { url: url.clone(), result: b.download() }
877                })
878                .collect()
879        })
880    }
881
882    /// Async version of [`download_all`](Self::download_all).
883    ///
884    /// Requires the `async` feature.  Uses `tokio::sync::Semaphore` for
885    /// concurrency control.
886    #[cfg(feature = "async")]
887    pub async fn download_all_async(self) -> Vec<BatchResult> {
888        use std::sync::Arc as StdArc;
889        use tokio::sync::Semaphore;
890        use tokio::task::spawn_blocking;
891
892        let semaphore = StdArc::new(Semaphore::new(self.concurrency));
893        let mut join_handles = Vec::new();
894
895        for url in self.urls.iter().cloned() {
896            let sem   = semaphore.clone();
897            let od    = self.output_dir.clone();
898            let pu    = self.proxy_url.clone();
899            let puser = self.proxy_user.clone();
900            let ppass = self.proxy_pass.clone();
901            let hdrs  = self.headers.clone();
902            let retry = self.retry.clone();
903            let sl    = self.speed_limit;
904            let quiet = self.quiet;
905
906            let permit = sem.acquire_owned().await.unwrap();
907            let h = spawn_blocking(move || {
908                let _permit = permit;
909                let filename = utils::get_filename_from_url_or_default(&url, "download");
910                let output_path = format!("{}/{}", od.trim_end_matches('/'), filename);
911
912                let mut b = DownloadBuilder::new(&url)
913                    .output(&output_path)
914                    .quiet(quiet)
915                    .retry(retry);
916                if let Some(lim) = sl { b = b.speed_limit(lim); }
917                if let Some(ref p) = pu { b = b.proxy(p.clone()); }
918                if let (Some(u), Some(p)) = (puser, ppass) {
919                    b = b.proxy_auth(u, p);
920                }
921                for (k, v) in hdrs { b = b.header(k, v); }
922
923                BatchResult { url, result: b.download() }
924            });
925            join_handles.push(h);
926        }
927
928        let mut results = Vec::new();
929        for h in join_handles {
930            if let Ok(r) = h.await { results.push(r); }
931        }
932        results
933    }
934}
935
936// ════════════════════════════════════════════════════════════════════════════
937// Free-function helpers
938// ════════════════════════════════════════════════════════════════════════════
939
940fn apply_headers(
941    mut req: reqwest::blocking::RequestBuilder,
942    headers: &[(String, String)],
943) -> reqwest::blocking::RequestBuilder {
944    for (name, value) in headers {
945        if let (Ok(n), Ok(v)) = (
946            reqwest::header::HeaderName::from_bytes(name.as_bytes()),
947            reqwest::header::HeaderValue::from_str(value),
948        ) {
949            req = req.header(n, v);
950        }
951    }
952    req
953}
954
955fn extract_percent(msg: &str) -> Option<f64> {
956    let (_, after) = msg.split_once("PROGRESS:")?;
957    let s = after.split('%').next()?.trim();
958    s.parse::<f64>().ok()
959}