rss_core 0.6.0

//! Module to handle input/output (i.e Download or recall imagery)
//! See [get method](QueryResult::get)

use crate::cache::FileCache;
use crate::query::QueryResult;
use crate::qvf::{QvfFilename, QvfFilenames};
use crate::utils::{Bbox, ImagerySource, run_gdal_command};
use anyhow::bail;
use anyhow::Context;
use anyhow::Result;
use chrono;
use async_process::Command as Async_Command;
use gdal::Dataset;
use log::{debug, info, warn};
use rayon::prelude::{IntoParallelIterator, ParallelIterator};
use serde_json::json;
use stac::{Asset, Item, ItemCollection};
use std::collections::HashMap;
use std::env;
use std::iter::FromIterator;
use std::path::PathBuf;
use std::process::Command;
use std::str::FromStr;
use std::sync::Arc;
use std::{thread, time};
use tokio::sync::Semaphore;
use tokio::task;

use reqwest::blocking::Client;
use serde_json::Value;
pub use stac;

/// Creates a copy of an STAC Item with replaced assets.
///
/// Replaces the duplicated `Item::new()` + field-by-field copy pattern found in
/// `get_remote()`, `get()`, `get_async()`, and `query_stac()`.
pub(crate) fn item_with_assets(item: Item, new_assets: HashMap<String, Asset>) -> Item {
    let mut f_item = Item::new(item.id);
    f_item.extensions = item.extensions;
    f_item.geometry = item.geometry;
    f_item.bbox = item.bbox;
    f_item.properties = item.properties;
    f_item.links = item.links;
    f_item.assets = new_assets;
    f_item.collection = item.collection;
    f_item.additional_fields = item.additional_fields;
    f_item
}

/// Returns crop window arguments for a GDAL command.
///
/// Replaces the duplicated `cmd.args([("-projwin"), ...])` pattern found in
/// `download_and_cache()`, `download_and_cache_async()`, `get()`, and `get_async()`.
fn crop_args(crop_window: Option<Bbox>, crop_window_epsg: Option<&str>) -> Vec<String> {
    let mut args = Vec::new();
    if let Some(w) = crop_window {
        args.extend_from_slice(&[
            "-projwin".to_string(),
            w.xmin.to_string(),
            w.ymax.to_string(),
            w.xmax.to_string(),
            w.ymin.to_string(),
        ]);
    }
    if let Some(w_epsg) = crop_window_epsg {
        args.extend_from_slice(&["-projwin_srs".to_string(), format!("EPSG:{w_epsg}")]);
    }
    args
}

fn configure_gdal_s3_defaults() {
    if env::var("AWS_NO_SIGN_REQUEST").is_err() {
        env::set_var("AWS_NO_SIGN_REQUEST", "YES");
    }
    if env::var("AWS_REGION").is_err() {
        env::set_var("AWS_REGION", "ap-southeast-2");
    }
}

/// Check if a file is cached and not expired.
///
/// Returns `Some(cache_path)` if the file is cached and valid, `None` otherwise.
fn is_cached(cache: &FileCache, vsi_url: &str) -> Option<std::path::PathBuf> {
    let cache_path = cache.cache_path(vsi_url);
    let meta_path = cache.meta_path(vsi_url);

    if !cache_path.exists() || !meta_path.exists() {
        return None;
    }

    let meta = std::fs::read_to_string(&meta_path).ok()?;
    let entry: crate::cache::file_cache::CacheEntry = serde_json::from_str(&meta).ok()?;
    let cached_at = chrono::DateTime::parse_from_rfc3339(&entry.cached_at).ok()?;
    let now = chrono::Utc::now();
    let elapsed = now.signed_duration_since(cached_at).to_std().ok()?;

    if elapsed <= cache.ttl() {
        Some(cache_path)
    } else {
        None
    }
}

/// Create a symlink from `link_path` to `target_path`.
fn create_symlink(target_path: &std::path::Path, link_path: &std::path::Path) -> Result<()> {
    if let Some(parent) = link_path.parent() {
        std::fs::create_dir_all(parent).context("Failed to create parent directory for symlink")?;
    }
    if link_path.exists() {
        std::fs::remove_file(link_path).context("Failed to remove existing file")?;
    }
    #[cfg(unix)]
    std::os::unix::fs::symlink(target_path, link_path)
        .context("Failed to create symlink")?;
    #[cfg(windows)]
    std::os::windows::fs::symlink_file(target_path, link_path)
        .context("Failed to create symlink")?;
    Ok(())
}

/// Download a file via gdal_translate and cache it, then create a symlink.
///
/// # Arguments
///
/// * `cache` - The file cache instance
/// * `vsi_url` - The VSI URL to download from
/// * `local_href` - The desired output path (will be a symlink)
/// * `crop_window` - Optional crop window
/// * `crop_window_epsg` - Optional EPSG code for crop window
fn download_and_cache(
    cache: &FileCache,
    vsi_url: &str,
    local_href: &std::path::Path,
    crop_window: Option<Bbox>,
    crop_window_epsg: Option<&str>,
) -> Result<()> {
    let cache_path = cache.cache_path(vsi_url);
    let meta_path = cache.meta_path(vsi_url);

    // Ensure cache directory exists
    if let Some(parent) = cache_path.parent() {
        std::fs::create_dir_all(parent).context("Failed to create cache subdirectory")?;
    }

    // Build gdal_translate command pointing to cache path
    let mut cmd = Command::new("gdal_translate");
    cmd.arg("-q");
    let ca = crop_args(crop_window, crop_window_epsg);
    if !ca.is_empty() {
        cmd.args(ca);
    }
    cmd.arg(vsi_url);
    cmd.arg(cache_path.to_str().context("Cache path not valid UTF-8")?);

    debug!("Cache miss, downloading {} to cache", vsi_url);
    cmd.spawn()
        .context("Failed to start gdal_translate for cache download")?
        .wait()
        .context("gdal_translate failed")?;

    // Write metadata
    let metadata = std::fs::metadata(&cache_path)
        .context("Failed to get cache file metadata")?;
    let entry = crate::cache::file_cache::CacheEntry {
        url: vsi_url.to_string(),
        cached_at: chrono::Utc::now().to_rfc3339(),
        size: metadata.len(),
    };
    let meta_json = serde_json::to_string(&entry).context("Failed to serialize cache entry")?;
    std::fs::write(&meta_path, meta_json).context("Failed to write cache metadata")?;

    // Create symlink from output to cache
    create_symlink(&cache_path, local_href)?;
    Ok(())
}

/// Async version of download_and_cache.
async fn download_and_cache_async(
    cache: &FileCache,
    vsi_url: &str,
    local_href: &std::path::Path,
    crop_window: Option<Bbox>,
    crop_window_epsg: Option<&str>,
) -> Result<()> {
    let cache_path = cache.cache_path(vsi_url);
    let meta_path = cache.meta_path(vsi_url);

    if let Some(parent) = cache_path.parent() {
        std::fs::create_dir_all(parent).context("Failed to create cache subdirectory")?;
    }

    let mut cmd = Async_Command::new("gdal_translate");
    cmd.arg("-q");
    let ca = crop_args(crop_window, crop_window_epsg);
    if !ca.is_empty() {
        cmd.args(ca);
    }
    cmd.arg(vsi_url);
    cmd.arg(cache_path.to_str().context("Cache path not valid UTF-8")?);

    debug!("Cache miss, downloading {} to cache (async)", vsi_url);
    cmd.output()
        .await
        .context("gdal_translate failed")?;

    let metadata = std::fs::metadata(&cache_path)
        .context("Failed to get cache file metadata")?;
    let entry = crate::cache::file_cache::CacheEntry {
        url: vsi_url.to_string(),
        cached_at: chrono::Utc::now().to_rfc3339(),
        size: metadata.len(),
    };
    let meta_json = serde_json::to_string(&entry).context("Failed to serialize cache entry")?;
    std::fs::write(&meta_path, meta_json).context("Failed to write cache metadata")?;

    create_symlink(&cache_path, local_href)?;
    Ok(())
}

/// Implement methods for query results.
impl QueryResult {
    /// Resolves the root VSI path for the current imagery source and collection.
    ///
    /// Replaces the duplicated `match self.source { ... }` pattern found in
    /// `get_remote()`, `get()`, and `get_async()`.
    fn root_vsi_path(&self) -> Result<&'static str> {
        match self.source {
            ImagerySource::Dea(_) => Ok("/vsis3/dea-public-data"),
            ImagerySource::PlanetaryComputer(_) => {
                Ok("/vsicurl/https://landsateuwest.blob.core.windows.net")
            }
            ImagerySource::Element(_) => {
                let r = match self.result.items.first().and_then(|i| i.collection.as_deref()) {
                    Some("sentinel-2-l2a") => {
                        "/vsicurl/https://sentinel-cogs.s3.us-west-2.amazonaws.com"
                    }
                    Some("landsat-c2-l2") => "/vsis3/usgs-landsat",
                    Some(other) => bail!("Unsupported collection: {}", other),
                    None => bail!("No collection on first item"),
                };
                Ok(r)
            }
            ImagerySource::Apollo(_) => bail!("Apollo does not support VSI root paths"),
        }
    }
    /// Filters both QueryResults to retain only items with matching datetime values.
    ///
    /// Iterates through both item collections and keeps only items whose datetime
    /// matches one present in the other collection. Modifies both `self` and `other`.
    pub fn filter_matching_dates(&mut self, other: &mut Self) -> Result<()> {
        // Collect dates from both self and other
        let self_dates: Vec<_> = self
            .result
            .items
            .iter()
            .filter_map(|i| i.properties.datetime)
            .collect();

        let other_dates: Vec<_> = other
            .result
            .items
            .iter()
            .filter_map(|i| i.properties.datetime)
            .collect();

        // Find intersection of dates
        let matching_dates: Vec<_> = self_dates
            .into_iter()
            .filter(|d| other_dates.contains(d))
            .collect();

        // Retain only items with matching dates in self
        self.result.items.retain(|item| {
            item.properties
                .datetime
                .is_some_and(|d| matching_dates.contains(&d))
        });

        // Retain only items with matching dates in other
        other.result.items.retain(|item| {
            item.properties
                .datetime
                .is_some_and(|d| matching_dates.contains(&d))
        });
        Ok(())
    }

    /// Returns a new QueryResult containing only items whose datetime matches
    /// an item in the provided `other` QueryResult.
    ///
    /// Asserts that the resulting item count matches `other`'s item count.
    ///
    /// # Example
    /// ```ignore
    /// let filtered = query.try_match_dates(&reference_query)?;
    /// ```
    pub fn try_match_dates(self, other: &Self) -> Result<Self> {
        let mut filtered_items = Vec::new();
        let items = self.result.clone();
        let other_dates: Vec<_> = other
            .result
            .items
            .iter()
            .map(|i| i.properties.datetime.unwrap())
            .collect();
        println!("other dates {:?}", other_dates);

        for item in items {
            let date = item.properties.datetime.unwrap();
            if other_dates.contains(&date) {
                filtered_items.push(item);
            }
        }

        let item_collection = ItemCollection::from_iter(filtered_items);
        let query_result = QueryResult {
            result: item_collection,
            source: self.source,
        };

        assert_eq!(&other.result.items.len(), &query_result.result.items.len());
        Ok(query_result)
    }
    /// Filters items by satellite platform name (e.g., "landsat8", "sentinel-2a").
    ///
    /// Returns a new QueryResult containing only items matching the specified platform.
    pub fn filter_platform(self, satelite_name: &str) -> Result<Self> {
        let mut filtered_items = Vec::new();
        let items = self.result.clone();
        for item in items {
            let platform = item.properties.additional_fields["platform"]
                .as_str()
                .unwrap();
            if platform == satelite_name {
                filtered_items.push(item);
            }
        }

        let filtered = ItemCollection::from_iter(filtered_items);
        let result = QueryResult {
            result: filtered,
            source: self.source,
        };
        Ok(result)
    }

    /// Returns an ItemCollection with remote URLs converted to local GDAL VSI paths.
    ///
    /// For DEA: uses `/vsis3/dea-public-data`
    /// For Element84 Sentinel2: uses `/vsicurl/https://sentinel-cogs.s3.us-west-2.amazonaws.com`
    /// For Element84 Landsat: uses `/vsis3/usgs-landsat`
    /// For PlanetaryComputer: uses `/vsicurl/https://landsateuwest.blob.core.windows.net`
    ///
    /// Does not download files, only transforms asset hrefs.
    pub fn get_remote(self) -> Result<ItemCollection> {
        let local_feature_collection = match self.source {
            ImagerySource::Apollo(_) => {
                bail!("Not supported")
            }
            ImagerySource::Dea(_)
            | ImagerySource::Element(_)
            | ImagerySource::PlanetaryComputer(_) => {
                let mut to_get = Vec::new();
                for item in &self.result.items {
                    let id = &item.id;
                    for (_, asset) in item.assets.clone() {
                        let href = asset.href;
                        to_get.push((id, href));
                    }
                }
                if !to_get.is_empty() {
                    info!("Getting metadata for {:?} items.", to_get.len());
                } else {
                    bail!("Nothing to get. Check your query!");
                }

                let root = self.root_vsi_path()?;
                //            let mut urls = Vec::new();
                //            to_get.into_iter().for_each(|(id, url)| {
                //                let loc = url.as_str().split("/").collect::<Vec<&str>>()[3..].join("/");
                //                let url = format!("{}/{}", root, loc);
                //                urls.push(url);
                // });
                let mut items = Vec::new();
                for item in self.result.items {
                    let mut local_assets: HashMap<String, Asset> = HashMap::new();
                    for (k, asset) in &item.assets {
                        let mut local_asset = asset.to_owned();
                        let loc =
                            asset.href.as_str().split('/').collect::<Vec<&str>>()[3..].join("/");
                        let url = format!("{}/{}", root, loc);
                        local_asset.href = url;
                        local_assets.insert(k.clone(), local_asset);
                    }
                    items.push(item_with_assets(item, local_assets));
                }

                ItemCollection::from_iter(items)
            }
        };
        Ok(local_feature_collection)
    }
    /// Get images from a query result.
    /// Example
    /// ```ignore
    /// use stac::ItemCollection;
    /// use rss_core::DEA;
    /// use rss_core::query::ImageQueryBuilder;
    /// use rss_core::qvf::Collection;
    /// use rss_core::utils::Intersects;
    /// use rss_core::utils::Cmp;
    /// use std::path::PathBuf;
    /// use chrono::NaiveDate;
    ///
    /// let source = DEA.clone();
    /// // This query will yield 52 Items (metadata only)
    /// let query = ImageQueryBuilder::new(
    ///    source,
    ///    Collection::Sentinel2,
    ///    Intersects::Scene(vec!["54kvu"]),
    /// )
    /// .bands(["nbart_red", "nbart_nir_1"])
    ///    .start_date(NaiveDate::parse_from_str("20210101", "%Y%m%d").unwrap())
    ///    .end_date(NaiveDate::parse_from_str("20210111", "%Y%m%d").unwrap())
    ///    .cloudcover((Cmp::Less, 5))
    ///    .build();
    /// let tmp_dir = PathBuf::from("/scratch/rsc8/hardtkel/tmp2");
    /// // Running the next line will download 52 images.
    /// // let feature_collection: ItemCollection = query
    /// //     .get(&tmp_dir, None, None)
    /// //     .expect("Could not download files");
    /// ```
    pub fn get(
        self,
        to: &PathBuf,
        crop_window: Option<Bbox>,
        crop_window_epsg: Option<&str>,
    ) -> Result<ItemCollection> {
        configure_gdal_s3_defaults();
        std::fs::create_dir(to).unwrap_or(());
        let local_feature_collection = match self.source {
            ImagerySource::Apollo(_) => {
                if let Some(_w) = crop_window {
                    unimplemented!("This hasn't been implemented yet.");
                }
                let mut to_recall = Vec::new();
                for item in &self.result.items {
                    for (_, asset) in item.assets.clone() {
                        let href = asset.href;
                        to_recall.push(QvfFilename::from_str(&href).unwrap());
                    }
                }

                if !to_recall.is_empty() {
                    let qvf_filenames = QvfFilenames {
                        qvf_filenames: to_recall,
                    };
                    let _ = qvf_filenames
                        .recall(to)
                        .expect("Could not recall the files");
                } else {
                    warn!("Noting to recall")
                }

                let mut items = Vec::new();
                for item in self.result.items {
                    let mut local_assets: HashMap<String, Asset> = HashMap::new();
                    debug!("Working on item: {:?}", item);
                    for multiband_asset in item.assets.values() {
                        let href_parts = multiband_asset.href.split('/').collect::<Vec<&str>>();
                        let file_name = href_parts.last().unwrap();
                        let multiband_filename = to.join(file_name);
                        let ds = Dataset::open(multiband_filename.clone())?;
                        let n_bands = ds.raster_count();
                        debug!("Asset has {} bands", n_bands);
                        let stem = multiband_filename.file_stem().unwrap().to_str().unwrap();
                        let _extension = multiband_filename.extension().unwrap().to_str().unwrap();
                        for band_id in 0..n_bands {
                            let single_band_stem = format!("{}_b{}", stem, band_id + 1);
                            let single_band = format!("{}.{}", single_band_stem, "tif");
                            let single_band =
                                multiband_filename.parent().unwrap().join(single_band);

                            let argv = &[
                                "gdal_translate",
                                "-b",
                                &format!("{}", band_id + 1),
                                "-q",
                                multiband_filename.to_str().unwrap(),
                                (single_band.to_str().unwrap()),
                            ];

                            run_gdal_command(argv);

                            let mut single_band_asset = multiband_asset.to_owned();
                            single_band_asset.href = single_band.to_str().unwrap().to_string();
                            //update additional fields/
                            let bn = band_id + 1;
                            let mut additional_fields = single_band_asset.additional_fields;
                            let asset_name =
                                format!("{}_{bn}", single_band_asset.title.clone().unwrap());
                            additional_fields.insert(
                                "eo:bands".to_string(),
                                json!([{"common_name": asset_name}]),
                            );
                            single_band_asset.additional_fields = additional_fields;
                            // insert to local assets HashMap!
                            local_assets.insert(asset_name, single_band_asset);
                        }
                    }
                    // for (k, asset) in &item.assets {
                    //     let mut local_asset = asset.to_owned();
                    //     let href_parts = asset.href.split('/').collect::<Vec<&str>>();
                    //     let file_name = href_parts.last().unwrap();
                    //     let local_href = to.join(file_name);
                    //     local_asset.href = local_href.to_str().unwrap().to_string();

                    //     //update additional fields/
                    //     let mut additional_fields = asset.additional_fields;
                    //     additional_fields.insert(
                    //        "eo:bands".to_string(),
                    //        json!([{"common_name": asset.title.unwrap()}]),);
                    //     local_asset.additional_fields = additional_fields;
                    //     // insert to local assets HashMap!
                    //     println!("Added: {local_asset:?}");
                    //     local_assets.insert(k.clone(), local_asset);
                    // }
                    items.push(item_with_assets(item, local_assets));
                }

                ItemCollection::from(items)
                // ItemCollection::from_iter(items)
            }
            ImagerySource::Dea(_)
            | ImagerySource::Element(_)
            | ImagerySource::PlanetaryComputer(_) => {
                let mut to_download = Vec::new();
                debug!("Items to download found {:?}", self.result.items);

                debug!("Items to download found {:?}", self.result.items);
                for item in &self.result.items {
                    let id = &item.id;
                    for (_, asset) in item.assets.clone() {
                        let href = asset.href;
                        to_download.push((id, href));
                    }
                }
                if !to_download.is_empty() {
                    info!("Downloading {:?} files.", to_download.len());

                    debug!("{:?}", to_download);
                } else {
                    info!("Nothing to download. Check your query!");
                }

                env::var("RSS_DOWNLOAD_THREADS").unwrap_or("8".to_string());

                // rayon::ThreadPoolBuilder::new().num_threads(n_threads)
                //  .build_global()
                //  .unwrap();
                let client = Client::new();

                let root = self.root_vsi_path()?;

                to_download.into_par_iter().for_each(|(id, url)| {
                    //let url = Url::parse(&url).unwrap();
                    let loc = url.as_str().split('/').collect::<Vec<&str>>()[3..].join("/");
                    //let root = root;
                    let mut url = format!("{}/{}", root, loc);
                    //let url = Url::parse(url_s).expect("Could not parse URL");
                    let href_parts = url.as_str().split('/').collect::<Vec<&str>>();
                    let file_name = href_parts.last().unwrap();
                    let parent = to.join(id);

                    debug!(
                        "Attempting to create {parent:?}"
                    );

                    //let vsicurl = format!("/vsicurl/{url}");
                    debug!("url {url:?}");
                    debug!("Parent {:?}", parent);
                    if !parent.exists(){
                        std::fs::create_dir_all(&parent).unwrap_or_else(|_| panic!("Could not create the parent folder {:?}",parent));
                    };
                    let local_href = parent.join(file_name);
                    let mut cmd = Command::new("gdal_translate");
                    cmd.arg("-q");
                    let ca = crop_args(crop_window, crop_window_epsg);
                    if !ca.is_empty() {
                        cmd.args(ca);
                    }
                    if let ImagerySource::PlanetaryComputer(_) = self.source {
                        let raw_url = url
    .strip_prefix("/vsicurl/")
    .unwrap_or(&url);

    debug!("Url to sign: {:?}", raw_url);
    // Sign the asset URL
    let signed_resp = client
        .get("https://planetarycomputer.microsoft.com/api/sas/v1/sign")
        .query(&[
            ("api-version", "2025-04-30-preview"),
            ("href", raw_url),
        ])
        .send()
        .unwrap();

    debug!("Signed {:?}", signed_resp);
    let body = signed_resp.text().unwrap();
let signed_json: Value = serde_json::from_str(&body).unwrap();


if let Some(signed_href) = signed_json["href"].as_str() {
    let signed_with_vsi = format!("/vsicurl/{}", signed_href);
    debug!("Signed {}: {}", raw_url, signed_with_vsi);
    url = signed_with_vsi;
} else {
    // Debug unexpected JSON (error details, etc.)
    debug!("Signing response did not contain href: {:?}", signed_json);
}

}

                    // Use cache to check/download/symlink
                    let cache = FileCache::from_env().unwrap_or_else(|_| {
                        FileCache::new(
                            std::path::PathBuf::from("/tmp/rss_cache"),
                            std::time::Duration::from_secs(604800),
                        ).unwrap()
                    });

                    // Check if the file is already present (symlink or regular file)
                    if local_href.exists() {
                        debug!("File already exists at {:?}", local_href);
                    } else if let Some(cached_path) = is_cached(&cache, &url) {
                        // Cache hit - create symlink
                        debug!("Cache hit for {}", url);
                        if let Err(e) = create_symlink(&cached_path, &local_href) {
                            warn!("Failed to create symlink: {}, falling back to download", e);
                            // Fall through to download
                            let _ = download_and_cache(&cache, &url, &local_href, crop_window, crop_window_epsg);
                        }
                    } else {
                        // Cache miss - download and cache
                        if let Err(e) = download_and_cache(&cache, &url, &local_href, crop_window, crop_window_epsg) {
                            warn!("Cache download failed for {:?}: {}, falling back to direct download", url, e);
                            // Fallback: direct download with retry
                            let mut was_downloaded = local_href.exists();
                            let mut n_retry = 1;
                            let max_retry = 20;
                            while (n_retry <= max_retry) && !was_downloaded {
                                if n_retry > 5 {
                                    let sleep_time = time::Duration::from_secs(10 * n_retry);
                                    thread::sleep(sleep_time);
                                }
                                let mut fallback_cmd = Command::new("gdal_translate");
                                fallback_cmd.arg("-q");
                                fallback_cmd.arg(&url);
                                fallback_cmd.arg(&local_href);
                                fallback_cmd
                                    .spawn()
                                    .expect("failed creating local file")
                                    .wait()
                                    .expect("failed to get the file");
                                was_downloaded = local_href.exists();
                                n_retry += 1;
                            }
                            if !was_downloaded {
                                panic!("The file {:?} could not be downloaded", url);
                            }
                        }
                    }
                });

                let mut items = Vec::new();
                for item in self.result.items {
                    let id = &item.id;
                    let mut local_assets: HashMap<String, Asset> = HashMap::new();
                    for (k, asset) in &item.assets {
                        let mut local_asset = asset.to_owned();
                        let href_parts = asset.href.split('/').collect::<Vec<&str>>();
                        let file_name = href_parts.last().unwrap();
                        let local_href = PathBuf::from(to).join(id).join(file_name);
                        local_asset.href = local_href.to_str().unwrap().to_string();
                        local_assets.insert(k.clone(), local_asset);
                    }
                    items.push(item_with_assets(item, local_assets));
                }

                ItemCollection::from_iter(items)
            }
        };
        Ok(local_feature_collection)
    }

    /// Async version of [`QueryResult::get`].
    ///
    /// Downloads files concurrently using a semaphore-bounded task pool.
    /// Supports the same crop window and EPSG parameters as the sync version.
    ///
    /// # Arguments
    ///
    /// * `to` - Destination directory
    /// * `n_downloads` - Maximum concurrent download tasks
    /// * `crop_window` - Optional bounding box for cropping
    /// * `crop_window_epsg` - Optional EPSG code for the crop window
    pub async fn get_async(
        self,
        to: &PathBuf,
        n_downloads: usize,
        crop_window: Option<Bbox>,
        crop_window_epsg: Option<&str>,
    ) -> Result<ItemCollection> {
        configure_gdal_s3_defaults();
        std::fs::create_dir(to).unwrap_or(());
        let semaphore = Arc::new(Semaphore::new(n_downloads));
        let local_feature_collection = match self.source {
            ImagerySource::Apollo(_) => {
                panic!("Async methods not supported in Apollo ");
            }
            ImagerySource::Dea(_)
            | ImagerySource::Element(_)
            | ImagerySource::PlanetaryComputer(_) => {
                let mut to_download = Vec::new();
                let result = self.result.clone();
                for item in result.items {
                    for (_, asset) in item.assets.clone() {
                        let href = asset.href;
                        to_download.push((item.id.clone(), href));
                    }
                }
                if !to_download.is_empty() {
                    info!("Downloading {:?} files.", to_download.len());
                    debug!("{:?}", to_download);
                } else {
                    info!("Nothing to download. Check your query!");
                }

                let root = self.root_vsi_path()?;

                let path = Arc::new(PathBuf::from(to).to_owned());
                let crop_window_epsg_arc = Arc::new(crop_window_epsg.map(|s| s.to_owned()));

                let overwrite = match env::var("RSS_OVERWRITE") {
                    Ok(value) => match value.as_str() {
                        "true" => true,
                        "false" => false,
                        _ => false,
                    },
                    Err(_) => false,
                };

                let tasks = to_download.into_iter().map(|(id, url)| {
                    let path_clone = Arc::clone(&path); // Create a new reference to the Arc inside the closure
                    let crop_window_epsg = Arc::clone(&crop_window_epsg_arc);
                    let semaphore = Arc::clone(&semaphore);
                    task::spawn(async move {
                        let permit = semaphore.clone().acquire_owned().await.unwrap();

                        let path_ref = PathBuf::from(path_clone.as_ref());
                        let loc = url.as_str().split('/').collect::<Vec<&str>>()[3..].join("/");
                        let url = &format!("{}/{}", root, loc);
                        let href_parts = url.as_str().split('/').collect::<Vec<&str>>();
                        let file_name = href_parts.last().unwrap();
                        let parent = path_ref.join(id);
                        debug!("Attempting to create {parent:?}");

                        debug!("url {url:?}");

                        std::fs::create_dir_all(&parent)
                            .expect("Could not create the parent folder");
                        let local_href = parent.join(file_name);

                        // Use cache for download
                        let cache = FileCache::from_env().unwrap_or_else(|_| {
                            FileCache::new(
                                std::path::PathBuf::from("/tmp/rss_cache"),
                                std::time::Duration::from_secs(604800),
                            ).unwrap()
                        });

                        let was_downloaded = local_href.exists();
                        let crop_epsg_str: Option<&str> = crop_window_epsg.as_ref().as_deref();
                        if !was_downloaded || overwrite {
                            if let Some(cached_path) = is_cached(&cache, url) {
                                // Cache hit - create symlink
                                debug!("Cache hit for {}", url);
                                if let Err(e) = create_symlink(&cached_path, &local_href) {
                                    warn!("Failed to create symlink: {}, falling back to download", e);
                                    let _ = download_and_cache_async(&cache, url, &local_href, crop_window, crop_epsg_str).await;
                                }
                            } else {
                                // Cache miss - download and cache
                                if let Err(e) = download_and_cache_async(&cache, url, &local_href, crop_window, crop_epsg_str).await {
                                    warn!("Async cache download failed for {:?}: {}, falling back", url, e);
                                    // Fallback: direct download
                                    let mut fallback_cmd = Async_Command::new("gdal_translate");
                                    fallback_cmd.arg("-q");
                                    if let Some(w) = crop_window {
                                        fallback_cmd.args([
                                            ("-projwin"),
                                            &format!("{}", w.xmin),
                                            &format!("{}", w.ymax),
                                            &format!("{}", w.xmax),
                                            &format!("{}", w.ymin),
                                        ]);
                                    }
                                    if let Some(w_epsg) = crop_window_epsg.as_ref() {
                                        fallback_cmd.args(["-projwin_srs", &format!("EPSG:{w_epsg}")]);
                                    }
                                    fallback_cmd.arg(url.clone())
                                        .arg(local_href.clone());
                                    let _ = fallback_cmd.output().await;
                                }
                            }
                            drop(permit);
                        };
                    })
                });
                let _ = futures::future::join_all(tasks).await;

                // for (id, url) in to_download {

                // to_download.into_par_iter().for_each(|(id, url)| {

                //     //let url = Url::parse(&url).unwrap();
                //     //let root = root;
                //     //let url = Url::parse(url_s).expect("Could not parse URL");
                //     let mut was_downloaded = local_href.exists() ; // check if file is alredy there...
                //     let mut n_retry = 1;
                //     let max_retry = 20;
                //     while (n_retry <= max_retry) & ! was_downloaded {
                //         if n_retry > 5 {
                //             let sleep_time = time::Duration::from_secs(10*n_retry);
                //             thread::sleep(sleep_time);
                //         }
                //         cmd.spawn()
                //             .expect("failed creating local file")
                //             .wait()
                //             .expect("failed to get the file");
                //         was_downloaded = local_href.exists();
                //         debug!("{n_retry:?} attempt to download {url:?} to {local_href:?} -> exist: {was_downloaded:?}");
                //         n_retry += 1;
                //     }
                //     if ! was_downloaded {
                //         panic!("The file {:?} could not be downloaded", url);
                //     }
                // });

                let mut items = Vec::new();
                for item in self.result.items {
                    let id = &item.id;
                    let mut local_assets: HashMap<String, Asset> = HashMap::new();
                    for (k, asset) in &item.assets {
                        let mut local_asset = asset.to_owned();
                        let href_parts = asset.href.split('/').collect::<Vec<&str>>();
                        let file_name = href_parts.last().unwrap();
                        let local_href = PathBuf::from(to).join(id).join(file_name);
                        local_asset.href = local_href.to_str().unwrap().to_string();
                        local_assets.insert(k.clone(), local_asset);
                    }
                    items.push(item_with_assets(item, local_assets));
                }

                ItemCollection::from_iter(items)
            }
        };
        Ok(local_feature_collection)
    }
}

#[cfg(test)]
mod tests {
    use crate::query::ImageQueryBuilder;
    use crate::qvf::Collection;
    use crate::utils::{Cmp, Intersects};
    use crate::{DEA, ELEMENT84, PLANETARYCOMPUTER};
    use chrono::NaiveDate;
    use std::path::PathBuf;
    use tempfile::TempDir;

    fn run_download_test(
        source: crate::utils::ImagerySource,
        collection: crate::qvf::Collection,
        scene: &str,
        band: &str,
    ) {
        let query =
            ImageQueryBuilder::new(source, collection, Intersects::Scene(vec![scene]))
                .bands([band])
                .start_date(NaiveDate::parse_from_str("20220101", "%Y%m%d").unwrap())
                .end_date(NaiveDate::parse_from_str("20220115", "%Y%m%d").unwrap())
                .cloudcover((Cmp::Less, 50))
                .build();

        let tmp_dir = TempDir::new().expect("Could not create temp directory");
        let result = query.get(&tmp_dir.path().to_path_buf(), None, None);

        assert!(result.is_ok(), "Query/get should succeed");
        let fc = result.unwrap();
        assert!(!fc.items.is_empty(), "Should have at least one item");

        let first_item = &fc.items[0];
        assert!(
            !first_item.assets.is_empty(),
            "First item should have assets"
        );

        let first_asset = first_item.assets.values().next().unwrap();
        let local_path = PathBuf::from(&first_asset.href);
        assert!(
            local_path.exists(),
            "Downloaded file should exist at: {}",
            first_asset.href
        );
    }

    #[test]
    fn test_dea_s3_download_single_band() {
        run_download_test(
            DEA.clone(),
            Collection::Sentinel2,
            "56jmr",
            "nbart_red",
        );
    }

    #[test]
    fn test_element84_sentinel2_download() {
        run_download_test(
            ELEMENT84.clone(),
            Collection::Sentinel2,
            "55kfp",
            "red",
        );
    }

    #[test]
    #[ignore = "request pays - requires credentials"]
    fn test_element84_landsat_download() {
        run_download_test(
            ELEMENT84.clone(),
            Collection::Landsat8,
            "p094r074",
            "red",
        );
    }

    #[test]
    fn test_planetary_computer_landsat_download() {
        run_download_test(
            PLANETARYCOMPUTER.clone(),
            Collection::Landsat8,
            "p094r074",
            "red",
        );
    }
}