//! Functions for publishing conda packages to various backends (local filesystem, S3, Quetz, etc.)

use miette::{Context, IntoDiagnostic};
use rattler_conda_types::{
    Channel, ChannelUrl, MatchSpec, NamedChannelOrUrl, PackageName, Platform,
};
use rattler_index::{IndexFsConfig, index_fs};
use rattler_repodata_gateway::{CacheClearMode, Gateway, SubdirSelection};
use std::collections::HashMap;
use std::path::{Path, PathBuf};

use crate::BuildString;
use crate::render::reporters::GatewayReporter;
use crate::tool_configuration::{self, Configuration};
use crate::types::Output;

/// Configuration for publishing packages to a channel.
/// This is the core-level type; CLI options are converted into this.
#[derive(Debug, Clone)]
pub struct PublishConfig {
    /// Whether to force overwrite existing packages
    pub force: bool,
    /// Whether to generate attestation for packages
    pub generate_attestation: bool,
    /// Authentication file path
    pub auth_file: Option<PathBuf>,
    /// S3 configuration
    #[cfg(feature = "s3")]
    pub s3_config: HashMap<String, rattler_networking::s3_middleware::S3Config>,
}

/// Represents a parsed build number argument
#[derive(Debug, Clone)]
pub enum BuildNumberOverride {
    /// Absolute build number (e.g., "12")
    Absolute(u64),
    /// Relative bump (e.g., "+1")
    Relative(i64),
}

impl BuildNumberOverride {
    /// Parse a build number string into either absolute or relative form
    pub fn parse(s: &str) -> miette::Result<Self> {
        let s = s.trim();
        if let Some(stripped) = s.strip_prefix('+') {
            let bump: i64 = stripped
                .parse()
                .map_err(|e| miette::miette!("Invalid relative build number '{}': {}", s, e))?;
            Ok(BuildNumberOverride::Relative(bump))
        } else if let Some(stripped) = s.strip_prefix('-') {
            let bump: i64 = stripped
                .parse::<i64>()
                .map_err(|e| miette::miette!("Invalid relative build number '{}': {}", s, e))?;
            Ok(BuildNumberOverride::Relative(-bump))
        } else {
            let num: u64 = s
                .parse()
                .map_err(|e| miette::miette!("Invalid absolute build number '{}': {}", s, e))?;
            Ok(BuildNumberOverride::Absolute(num))
        }
    }
}

/// Fetch the highest build number for packages from the target channel
pub async fn fetch_highest_build_numbers(
    target_url: &NamedChannelOrUrl,
    outputs: &[Output],
    target_platform: Platform,
    tool_config: &Configuration,
) -> miette::Result<HashMap<(PackageName, String), u64>> {
    // Convert target URL to channel
    let channel = match target_url {
        NamedChannelOrUrl::Url(url) => Channel::from_url(ChannelUrl::from(url.clone())),
        NamedChannelOrUrl::Path(path) => {
            let url = url::Url::from_file_path(path.as_str())
                .map_err(|_| miette::miette!("Invalid path: {}", path))?;
            Channel::from_url(ChannelUrl::from(url))
        }
        NamedChannelOrUrl::Name(name) => {
            return Err(miette::miette!(
                "Cannot fetch repodata from named channel '{}'. Please use a URL.",
                name
            ));
        }
    };

    // Collect unique package names from outputs (we'll filter by version later)
    let mut package_specs: Vec<MatchSpec> = Vec::new();
    let mut versions_to_check: HashMap<PackageName, Vec<String>> = HashMap::new();

    for output in outputs {
        let name = output.name().clone();
        let version = output.recipe.package().version().to_string();

        // Track versions we're interested in
        versions_to_check
            .entry(name.clone())
            .or_default()
            .push(version);

        // Create a matchspec that matches the package name (any version)
        let spec = MatchSpec {
            name: rattler_conda_types::PackageNameMatcher::Exact(name),
            ..Default::default()
        };
        if !package_specs.iter().any(|s| s.name == spec.name) {
            package_specs.push(spec);
        }
    }

    if package_specs.is_empty() {
        return Ok(HashMap::new());
    }

    let span = tracing::info_span!("Fetching build numbers from target channel",);
    let _guard = span.enter();

    // Query the repodata
    let result = tool_config
        .repodata_gateway
        .query(
            vec![channel],
            [target_platform, Platform::NoArch],
            package_specs,
        )
        .with_reporter(
            GatewayReporter::builder()
                .with_multi_progress(tool_config.fancy_log_handler.multi_progress().clone())
                .with_progress_template(tool_config.fancy_log_handler.default_bytes_style())
                .with_finish_template(tool_config.fancy_log_handler.finished_progress_style())
                .finish(),
        )
        .recursive(false)
        .await;

    tool_config
        .fancy_log_handler
        .multi_progress()
        .clear()
        .unwrap();

    // Process results to find highest build numbers
    let mut highest_build_numbers: HashMap<(PackageName, String), u64> = HashMap::new();

    match result {
        Ok(repo_data) => {
            // Surface any non-fatal warnings gathered during the query
            // instead of dropping them.
            for warning in &repo_data.warnings {
                tracing::warn!("{warning}");
            }
            for repo in repo_data {
                for record in repo.iter() {
                    let name = &record.package_record.name;
                    let version = record.package_record.version.version().to_string();

                    // Only track versions we're actually building
                    if let Some(versions) = versions_to_check.get(name)
                        && versions.contains(&version)
                    {
                        let key = (name.clone(), version);
                        let build_number = record.package_record.build_number;
                        highest_build_numbers
                            .entry(key)
                            .and_modify(|e| *e = (*e).max(build_number))
                            .or_insert(build_number);
                    }
                }
            }
        }
        Err(e) => {
            // Log the error but don't fail - the channel might not exist yet or be empty
            tracing::debug!("Could not fetch repodata from target channel: {}", e);
        }
    }

    Ok(highest_build_numbers)
}

/// Apply build number override to outputs
pub fn apply_build_number_override(
    outputs: &mut [Output],
    build_number_override: &BuildNumberOverride,
    highest_build_numbers: &HashMap<(PackageName, String), u64>,
) {
    let span = tracing::info_span!("Applying build number overrides",);
    let _guard = span.enter();
    for output in outputs {
        let name = output.name().clone();
        let version = output.recipe.package().version().to_string();
        let key = (name.clone(), version.clone());

        let new_build_number = match build_number_override {
            BuildNumberOverride::Absolute(num) => *num,
            BuildNumberOverride::Relative(bump) => {
                let current_highest = highest_build_numbers.get(&key).copied().unwrap_or(0);
                let new_num = (current_highest as i64 + bump).max(0) as u64;
                tracing::info!(
                    "Packaging {} ({}): bumping build number from {} to {} ({}{})",
                    name.as_normalized(),
                    version,
                    current_highest,
                    new_num,
                    if *bump >= 0 { "+" } else { "" },
                    bump
                );
                new_num
            }
        };

        // Update the build number
        output.recipe.build.number = Some(new_build_number);

        // Extract the hash from the current build string and recompute with new build number
        let current_build_string = output
            .recipe
            .build
            .string
            .as_resolved()
            .expect("Build string should be resolved at this point");

        // Split on last '_' to separate hash from build number
        if let Some(last_underscore) = current_build_string.rfind('_') {
            let hash_part = &current_build_string[..last_underscore];
            let new_build_string = format!("{}_{}", hash_part, new_build_number);
            output.recipe.build.string = BuildString::Resolved(new_build_string);
        }
    }
}

/// Helper function to determine the package subdirectory (platform)
pub fn determine_package_subdir(package_path: &Path) -> miette::Result<String> {
    use rattler_conda_types::package::IndexJson;
    use rattler_package_streaming::seek::read_package_file;

    let index_json: IndexJson = read_package_file(package_path)
        .map_err(|e| miette::miette!("Failed to read package file: {}", e))?;

    Ok(index_json.subdir.unwrap_or_else(|| "noarch".to_string()))
}

/// Upload packages to a channel and run indexing.
/// After the indexing, the repodata cache for the target channel is cleared.
pub async fn upload_and_index_channel(
    target_url: &NamedChannelOrUrl,
    package_paths: &[PathBuf],
    publish_config: &PublishConfig,
    repodata_gateway: &Gateway,
) -> miette::Result<()> {
    let span = tracing::info_span!("Publishing packages");
    let _guard = span.enter();

    // Collect subdirs from packages for cache clearing later
    let subdirs: std::collections::HashSet<String> = package_paths
        .iter()
        .filter_map(|p| determine_package_subdir(p).ok())
        .collect();

    match target_url {
        NamedChannelOrUrl::Url(url) => {
            let scheme = url.scheme();

            match scheme {
                "s3" => {
                    #[cfg(not(feature = "s3"))]
                    {
                        return Err(miette::miette!(
                            "S3 support is not enabled. Please recompile with the 's3' feature."
                        ));
                    }

                    #[cfg(feature = "s3")]
                    {
                        upload_to_s3(url, package_paths, publish_config).await
                    }
                }
                "quetz" => upload_to_quetz(url, package_paths, publish_config).await,
                "artifactory" => upload_to_artifactory(url, package_paths, publish_config).await,
                "prefix" => upload_to_prefix(url, package_paths, publish_config).await,
                "cloudsmith" => upload_to_cloudsmith(url, package_paths, publish_config).await,
                "file" => {
                    let path = url
                        .to_file_path()
                        .map_err(|()| miette::miette!("Invalid file URL: {}", url))?;
                    upload_to_local_filesystem(&path, package_paths, publish_config.force).await
                }
                "http" | "https" => {
                    // Detect backend from hostname
                    let host = url.host_str().unwrap_or("");

                    if host.contains("prefix.dev") {
                        upload_to_prefix(url, package_paths, publish_config).await
                    } else if host.contains("anaconda.org") {
                        upload_to_anaconda(url, package_paths, publish_config).await
                    } else if host.contains("quetz") {
                        upload_to_quetz(url, package_paths, publish_config).await
                    } else {
                        Err(miette::miette!(
                            "Cannot determine upload backend from URL '{}'. \n\
                            Supported hosts: prefix.dev, anaconda.org, or use explicit schemes: s3://, quetz://, artifactory://, prefix://, cloudsmith://",
                            url
                        ))
                    }
                }
                _ => Err(miette::miette!(
                    "Unsupported URL scheme '{}'. Supported schemes: file://, s3://, quetz://, artifactory://, prefix://, cloudsmith://, http://, https://",
                    scheme
                )),
            }
        }
        NamedChannelOrUrl::Path(path) => {
            let path_buf = PathBuf::from(path.as_str());
            upload_to_local_filesystem(&path_buf, package_paths, publish_config.force).await
        }
        NamedChannelOrUrl::Name(name) => Err(miette::miette!(
            "Cannot upload to named channel '{}'. Please use a direct URL instead.",
            name
        )),
    }?;

    // Clear repodata cache for the target channel after publishing. Resolve the target
    // first so schemes the gateway can't read (e.g. `cloudsmith://`) map to the same
    // channel URL that was used while building, and the correct cache is invalidated.
    let channel = match resolve_channel_for_repodata(target_url) {
        NamedChannelOrUrl::Url(url) => Channel::from_url(ChannelUrl::from(url)),
        NamedChannelOrUrl::Path(path) => {
            let url = url::Url::from_file_path(path.as_str())
                .map_err(|_| miette::miette!("Invalid path: {}", path))?;
            Channel::from_url(ChannelUrl::from(url))
        }
        NamedChannelOrUrl::Name(_) => {
            // Named channels are already rejected above, so this is unreachable
            unreachable!()
        }
    };

    repodata_gateway
        .clear_repodata_cache(
            &channel,
            SubdirSelection::Some(subdirs),
            CacheClearMode::InMemoryAndDisk,
        )
        .into_diagnostic()?;

    tracing::debug!("Cleared repodata cache for target channel");

    Ok(())
}

/// Default host serving Cloudsmith conda channels. Override with `CLOUDSMITH_CHANNEL_HOST`.
const DEFAULT_CLOUDSMITH_CHANNEL_HOST: &str = "conda.cloudsmith.io";

/// Parse a `cloudsmith://owner/repo` URL into its `(owner, repo)` components.
///
/// Returns `None` unless the URL has exactly an owner (host) and a single non-empty
/// repo path segment, letting callers decide how to handle a malformed target.
fn parse_cloudsmith_owner_repo(url: &url::Url) -> Option<(String, String)> {
    let owner = url.host_str()?.to_string();
    let mut segments = url
        .path_segments()
        .map(|s| s.filter(|seg| !seg.is_empty()))?;
    let repo = segments.next()?.to_string();
    if segments.next().is_some() {
        return None;
    }
    Some((owner, repo))
}

/// Resolve a publish target to the channel URL used for repodata operations: build-time
/// dependency resolution, build-number lookups, and repodata cache clearing.
///
/// The repodata resolver can't read the `cloudsmith://` scheme, so a
/// `cloudsmith://owner/repo` target is rewritten to `https://<host>/owner/repo` (host from
/// `CLOUDSMITH_CHANNEL_HOST`, default [`DEFAULT_CLOUDSMITH_CHANNEL_HOST`]). The original
/// `cloudsmith://` target is kept as the upload target elsewhere so upload dispatch stays
/// host-independent. Any target that isn't a well-formed `cloudsmith://owner/repo` URL is
/// returned unchanged.
pub fn resolve_channel_for_repodata(target: &NamedChannelOrUrl) -> NamedChannelOrUrl {
    let NamedChannelOrUrl::Url(url) = target else {
        return target.clone();
    };
    if url.scheme() != "cloudsmith" {
        return target.clone();
    }
    let Some((owner, repo)) = parse_cloudsmith_owner_repo(url) else {
        return target.clone();
    };

    let host = std::env::var("CLOUDSMITH_CHANNEL_HOST")
        .unwrap_or_else(|_| DEFAULT_CLOUDSMITH_CHANNEL_HOST.to_string());

    match url::Url::parse(&format!(
        "https://{}/{owner}/{repo}",
        host.trim_end_matches('/')
    )) {
        Ok(converted) => NamedChannelOrUrl::Url(converted),
        Err(_) => target.clone(),
    }
}

/// Upload packages to Cloudsmith.
async fn upload_to_cloudsmith(
    url: &url::Url,
    package_paths: &[PathBuf],
    publish_config: &PublishConfig,
) -> miette::Result<()> {
    use rattler_upload::upload::opt::CloudsmithData;
    use rattler_upload::upload::upload_package_to_cloudsmith;

    tracing::info!("Uploading packages to Cloudsmith: {}", url);

    let (owner, repo) = parse_cloudsmith_owner_repo(url).ok_or_else(|| {
        miette::miette!(
            "Invalid Cloudsmith URL '{}': expected cloudsmith://owner/repo",
            url
        )
    })?;

    let api_key = std::env::var("CLOUDSMITH_API_KEY").ok();
    let api_url = std::env::var("CLOUDSMITH_API_URL")
        .ok()
        .map(|url| url.parse())
        .transpose()
        .into_diagnostic()
        .context("Failed to parse CLOUDSMITH_API_URL")?;
    let cloudsmith_data = CloudsmithData::new(owner, repo, api_key, api_url);

    let auth_storage = tool_configuration::get_auth_store(publish_config.auth_file.clone())
        .map_err(|e| miette::miette!("Failed to get authentication storage: {}", e))?;

    upload_package_to_cloudsmith(&auth_storage, &package_paths.to_vec(), cloudsmith_data)
        .await
        .into_diagnostic()
        .context("Failed to upload packages to Cloudsmith")
}

#[cfg(feature = "s3")]
/// Upload packages to S3 and run indexing
async fn upload_to_s3(
    url: &url::Url,
    package_paths: &[PathBuf],
    publish_config: &PublishConfig,
) -> miette::Result<()> {
    use rattler_index::{IndexS3Config, ensure_channel_initialized_s3, index_s3};
    use rattler_upload::upload::upload_package_to_s3;
    use std::collections::HashSet;

    tracing::info!("Uploading packages to S3 channel: {}", url);

    // Resolve S3 credentials using config + auth storage, falling back to AWS SDK
    let resolved_credentials = tool_configuration::resolve_s3_credentials(
        &publish_config.s3_config,
        publish_config.auth_file.clone(),
        url,
    )
    .await
    .into_diagnostic()?;

    // Ensure channel is initialized with noarch/repodata.json
    ensure_channel_initialized_s3(url, &resolved_credentials)
        .await
        .map_err(|e| miette::miette!("Failed to initialize S3 channel: {}", e))?;

    // Collect unique subdirs from all packages
    let mut subdirs = HashSet::new();
    for package_path in package_paths {
        let subdir = determine_package_subdir(package_path)?;
        subdirs.insert(subdir);
    }

    // Upload packages to S3
    upload_package_to_s3(
        url.clone(),
        resolved_credentials.clone(),
        package_paths,
        publish_config.force,
    )
    .await
    .map_err(|e| miette::miette!("Failed to upload packages to S3: {}", e))?;

    tracing::info!("Successfully uploaded packages to S3");

    for subdir in subdirs {
        // Run S3 indexing for each subdir
        tracing::info!("Indexing S3 channel at {} / {}", url, subdir);

        let target_platform = subdir
            .parse::<Platform>()
            .map_err(|e| miette::miette!("Invalid platform subdir '{}': {}", subdir, e))?;

        let index_config = IndexS3Config {
            channel: url.clone(),
            credentials: resolved_credentials.clone(),
            target_platform: Some(target_platform),
            repodata_patch: None,
            write_zst: true,
            write_shards: true,
            repodata_revisions: Vec::new(),
            package_revision_assignment: Default::default(),
            force: false,
            max_parallel: num_cpus::get_physical(),
            multi_progress: None,
            precondition_checks: rattler_index::PreconditionChecks::Enabled,
        };

        index_s3(index_config)
            .await
            .map_err(|e| miette::miette!("Failed to index S3 channel: {}", e))?;
    }

    tracing::info!("Successfully indexed S3 channel");
    Ok(())
}

/// Upload packages to Quetz server
async fn upload_to_quetz(
    url: &url::Url,
    package_paths: &[PathBuf],
    publish_config: &PublishConfig,
) -> miette::Result<()> {
    use rattler_upload::upload::opt::QuetzData;
    use rattler_upload::upload::upload_package_to_quetz;

    tracing::info!("Uploading packages to Quetz server: {}", url);

    // Get authentication storage
    let auth_storage = tool_configuration::get_auth_store(publish_config.auth_file.clone())
        .map_err(|e| miette::miette!("Failed to get authentication storage: {}", e))?;

    // Extract channel name from URL path
    let channel = url
        .path_segments()
        .and_then(|mut segments| segments.next_back())
        .ok_or_else(|| miette::miette!("Invalid Quetz URL: missing channel name"))?
        .to_string();

    // Convert quetz:// to https:// if needed, and strip the channel path from the URL
    // The server_url should only contain the base URL (e.g., https://quetz.example.com/)
    // without the channel path, since the channel is passed separately to QuetzData
    let mut server_url = url.clone();
    if server_url.scheme() == "quetz" {
        server_url
            .set_scheme("https")
            .map_err(|_| miette::miette!("Failed to convert quetz:// URL to https://"))?;
    }
    // Remove the channel path from the URL to get just the base server URL
    server_url.set_path("");

    // Create QuetzData with server URL, channel, and optional API key
    let quetz_data = QuetzData::new(server_url, channel, None);

    // Upload packages
    upload_package_to_quetz(&auth_storage, &package_paths.to_vec(), quetz_data)
        .await
        .map_err(|e| miette::miette!("Failed to upload packages to Quetz: {}", e))?;

    tracing::info!("Successfully uploaded packages to Quetz");
    tracing::info!("Note: Quetz handles indexing automatically on the server side");
    Ok(())
}

/// Upload packages to Artifactory server
async fn upload_to_artifactory(
    url: &url::Url,
    package_paths: &[PathBuf],
    publish_config: &PublishConfig,
) -> miette::Result<()> {
    use rattler_upload::upload::opt::ArtifactoryData;
    use rattler_upload::upload::upload_package_to_artifactory;

    tracing::info!("Uploading packages to Artifactory server: {}", url);

    // Get authentication storage
    let auth_storage = tool_configuration::get_auth_store(publish_config.auth_file.clone())
        .map_err(|e| miette::miette!("Failed to get authentication storage: {}", e))?;

    // Extract channel name from URL path
    let channel = url
        .path_segments()
        .and_then(|mut segments| segments.next_back())
        .ok_or_else(|| miette::miette!("Invalid Artifactory URL: missing repository name"))?
        .to_string();

    // Convert artifactory:// to https:// if needed, and strip the channel path from the URL
    // The server_url should only contain the base URL (e.g., https://artifactory.example.com/)
    // without the channel path, since the channel is passed separately to ArtifactoryData
    let mut server_url = url.clone();
    if server_url.scheme() == "artifactory" {
        server_url
            .set_scheme("https")
            .map_err(|_| miette::miette!("Failed to convert artifactory:// URL to https://"))?;
    }
    // Remove the channel path from the URL to get just the base server URL
    server_url.set_path("");

    // Create ArtifactoryData with server URL and channel; credentials are
    // resolved from the auth storage during upload
    let artifactory_data = ArtifactoryData::new(server_url, channel);

    // Upload packages
    upload_package_to_artifactory(&auth_storage, &package_paths.to_vec(), artifactory_data)
        .await
        .map_err(|e| miette::miette!("Failed to upload packages to Artifactory: {}", e))?;

    tracing::info!("Successfully uploaded packages to Artifactory");
    tracing::info!("Note: Artifactory handles indexing automatically on the server side");
    Ok(())
}

/// Upload packages to Prefix.dev server
async fn upload_to_prefix(
    url: &url::Url,
    package_paths: &[PathBuf],
    publish_config: &PublishConfig,
) -> miette::Result<()> {
    use rattler_upload::upload::opt::{
        AttestationSource, ForceOverwrite, PrefixData, SkipExisting,
    };
    use rattler_upload::upload::upload_package_to_prefix;

    tracing::info!("Uploading packages to Prefix.dev server: {}", url);

    // Get authentication storage
    let auth_storage = tool_configuration::get_auth_store(publish_config.auth_file.clone())
        .map_err(|e| miette::miette!("Failed to get authentication storage: {}", e))?;

    // Extract channel name from URL path
    let channel = url
        .path_segments()
        .and_then(|mut segments| segments.next_back())
        .ok_or_else(|| miette::miette!("Invalid Prefix URL: missing channel name"))?
        .to_string();

    // Convert prefix:// to https:// if needed, and strip the channel path from the URL
    // The server_url should only contain the base URL (e.g., https://prefix.dev/)
    // without the channel path, since the channel is passed separately to PrefixData
    let mut server_url = url.clone();
    if server_url.scheme() == "prefix" {
        server_url
            .set_scheme("https")
            .map_err(|_| miette::miette!("Failed to convert prefix:// URL to https://"))?;
    }
    // Remove the channel path from the URL to get just the base server URL
    server_url.set_path("");

    // Determine attestation source
    let attestation = if publish_config.generate_attestation {
        AttestationSource::GenerateAttestation
    } else {
        AttestationSource::NoAttestation
    };

    // Create PrefixData with server URL, channel, optional API key, attestation, skip_existing and force
    let prefix_data = PrefixData::new(
        server_url,
        channel,
        None,
        attestation,
        SkipExisting(false),
        ForceOverwrite(publish_config.force),
        false, // store_github_attestation
    );

    // Upload packages
    upload_package_to_prefix(&auth_storage, &package_paths.to_vec(), prefix_data)
        .await
        .map_err(|e| miette::miette!("Failed to upload packages to Prefix: {}", e))?;

    tracing::info!("Successfully uploaded packages to Prefix.dev");
    tracing::info!("Note: Prefix.dev handles indexing automatically on the server side");
    Ok(())
}

/// Upload packages to Anaconda.org
async fn upload_to_anaconda(
    url: &url::Url,
    package_paths: &[PathBuf],
    publish_config: &PublishConfig,
) -> miette::Result<()> {
    use rattler_upload::upload::opt::{AnacondaData, ForceOverwrite};
    use rattler_upload::upload::upload_package_to_anaconda;

    tracing::info!("Uploading packages to Anaconda.org: {}", url);

    // Get authentication storage
    let auth_storage = tool_configuration::get_auth_store(publish_config.auth_file.clone())
        .map_err(|e| miette::miette!("Failed to get authentication storage: {}", e))?;

    // Parse URL path to extract owner and optional channel
    // Expected format: https://anaconda.org/owner/channel or https://anaconda.org/owner
    let path_segments: Vec<&str> = url
        .path_segments()
        .ok_or_else(|| miette::miette!("Invalid Anaconda.org URL: missing path"))?
        .collect();

    let (owner, channel) = match path_segments.len() {
        1 => (path_segments[0].to_string(), "main".to_string()),
        2 => (path_segments[0].to_string(), path_segments[1].to_string()),
        _ => {
            return Err(miette::miette!(
                "Invalid Anaconda.org URL format. Expected: https://anaconda.org/owner or https://anaconda.org/owner/label"
            ));
        }
    };

    let anaconda_data = AnacondaData::new(
        owner,
        Some(vec![channel]),
        None,
        None,
        ForceOverwrite(publish_config.force),
    );

    upload_package_to_anaconda(&auth_storage, &package_paths.to_vec(), anaconda_data)
        .await
        .into_diagnostic()?;

    tracing::info!("Successfully uploaded packages to Anaconda.org");
    tracing::info!("Note: Anaconda.org handles indexing automatically on the server side");
    Ok(())
}

/// Upload packages to local filesystem and run indexing
async fn upload_to_local_filesystem(
    target_dir: &Path,
    package_paths: &[PathBuf],
    force: bool,
) -> miette::Result<()> {
    use rattler_index::ensure_channel_initialized_fs;
    use std::collections::HashSet;

    tracing::info!(
        "Copying packages to local channel: {}",
        target_dir.display()
    );

    // Create target directory if it doesn't exist
    fs_err::create_dir_all(target_dir).into_diagnostic()?;

    // Ensure channel is initialized with noarch/repodata.json
    ensure_channel_initialized_fs(target_dir)
        .await
        .map_err(|e| miette::miette!("Failed to initialize local channel: {}", e))?;

    // Collect unique subdirs from all packages
    let mut subdirs = HashSet::new();

    // Copy packages to the target directory organized by platform
    for package_path in package_paths {
        // Extract platform from package filename or metadata
        let package_name = package_path
            .file_name()
            .ok_or_else(|| miette::miette!("Invalid package path"))?;

        // Determine subdir from package
        let subdir = determine_package_subdir(package_path)?;
        subdirs.insert(subdir.clone());
        let target_subdir = target_dir.join(&subdir);

        fs_err::create_dir_all(&target_subdir).into_diagnostic()?;
        let target_path = target_subdir.join(package_name);

        // Check if package already exists
        if target_path.exists() && !force {
            return Err(miette::miette!(
                "Package already exists at {}. Use --force to overwrite.",
                target_path.display()
            ));
        }

        tracing::info!(
            "Copying {} to {}",
            package_path.display(),
            target_path.display()
        );
        fs_err::copy(package_path, &target_path).into_diagnostic()?;
    }

    // Run rattler-index on the local directory for each subdir
    tracing::info!("Indexing local channel at {}", target_dir.display());

    for subdir in subdirs {
        let target_platform = subdir
            .parse::<Platform>()
            .map_err(|e| miette::miette!("Invalid platform subdir '{}': {}", subdir, e))?;

        let index_config = IndexFsConfig {
            channel: target_dir.to_path_buf(),
            target_platform: Some(target_platform),
            repodata_patch: None,
            write_zst: true,
            write_shards: true,
            repodata_revisions: Vec::new(),
            package_revision_assignment: Default::default(),
            force: false,
            max_parallel: num_cpus::get_physical(),
            multi_progress: None,
        };

        index_fs(index_config)
            .await
            .map_err(|e| miette::miette!("Failed to index channel: {}", e))?;
    }

    tracing::info!("Successfully indexed local channel");
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use serial_test::serial;
    use std::str::FromStr;

    fn resolve(raw: &str) -> NamedChannelOrUrl {
        resolve_channel_for_repodata(&NamedChannelOrUrl::from_str(raw).unwrap())
    }

    #[test]
    fn parse_cloudsmith_owner_repo_extracts_components() {
        let url = url::Url::parse("cloudsmith://my-org/my-repo").unwrap();
        assert_eq!(
            parse_cloudsmith_owner_repo(&url),
            Some(("my-org".to_string(), "my-repo".to_string()))
        );
        // Missing repo or an extra segment is not a valid owner/repo pair.
        assert_eq!(
            parse_cloudsmith_owner_repo(&url::Url::parse("cloudsmith://my-org").unwrap()),
            None
        );
        assert_eq!(
            parse_cloudsmith_owner_repo(
                &url::Url::parse("cloudsmith://my-org/my-repo/extra").unwrap()
            ),
            None
        );
    }

    #[test]
    #[serial]
    fn resolve_channel_rewrites_cloudsmith_with_default_host() {
        unsafe { std::env::remove_var("CLOUDSMITH_CHANNEL_HOST") };
        assert_eq!(
            resolve("cloudsmith://my-org/my-repo"),
            NamedChannelOrUrl::Url(
                url::Url::parse("https://conda.cloudsmith.io/my-org/my-repo").unwrap()
            )
        );
    }

    #[test]
    #[serial]
    fn resolve_channel_honors_custom_host_and_trailing_slash() {
        unsafe { std::env::set_var("CLOUDSMITH_CHANNEL_HOST", "conda.example.com/") };
        let resolved = resolve("cloudsmith://my-org/my-repo");
        unsafe { std::env::remove_var("CLOUDSMITH_CHANNEL_HOST") };
        assert_eq!(
            resolved,
            NamedChannelOrUrl::Url(
                url::Url::parse("https://conda.example.com/my-org/my-repo").unwrap()
            )
        );
    }

    // Malformed cloudsmith targets short-circuit before the env var is read, so they need
    // no `#[serial]`. They are passed through unchanged and rejected later by the upload.
    #[test]
    fn resolve_channel_passes_through_malformed_cloudsmith_targets() {
        for raw in ["cloudsmith://my-org/my-repo/extra", "cloudsmith://my-org"] {
            assert_eq!(
                resolve(raw),
                NamedChannelOrUrl::from_str(raw).unwrap(),
                "{raw} should be passed through unchanged"
            );
        }
    }

    #[test]
    fn resolve_channel_leaves_other_targets_untouched() {
        for raw in [
            "https://prefix.dev/my-channel",
            "s3://my-bucket",
            "conda-forge",
        ] {
            assert_eq!(
                resolve(raw),
                NamedChannelOrUrl::from_str(raw).unwrap(),
                "{raw} should be unchanged"
            );
        }
    }
}
