use std::collections::BTreeSet;
use std::path::Path;
use std::time::Instant;

use crate::formatters::format_duration;
use crate::logline;
use crate::ui::UiTheme;
use surge_core::config::constants::{DEFAULT_ZSTD_LEVEL, RELEASES_FILE_COMPRESSED};
use surge_core::config::manifest::SurgeManifest;
use surge_core::error::{Result, SurgeError};
use surge_core::releases::manifest::{compress_release_index, decompress_release_index};
use surge_core::releases::restore::required_artifacts_for_index;
use surge_core::releases::version::compare_versions;
use surge_core::storage;

/// Compact a channel to a single latest full release and prune stale artifacts.
///
/// When `app_id` and `rid` are omitted, iterates over every app and target in the manifest.
pub async fn execute(
    manifest_path: &Path,
    app_id: Option<&str>,
    rid: Option<&str>,
    channel: Option<&str>,
) -> Result<()> {
    let manifest = SurgeManifest::from_file(manifest_path)?;
    let requested_channel = channel.map(str::trim).filter(|value| !value.is_empty());

    let targets: Vec<(String, String)> = if let Some(app_id) = app_id {
        let app_id = app_id.to_string();
        let rid = super::resolve_rid(&manifest, &app_id, rid)?;
        vec![(app_id, rid)]
    } else {
        manifest
            .app_ids()
            .into_iter()
            .flat_map(|app_id| {
                manifest
                    .target_rids(&app_id)
                    .into_iter()
                    .map(move |rid| (app_id.clone(), rid))
            })
            .collect()
    };
    let mut prepared_targets = Vec::with_capacity(targets.len());
    let mut preflight_errors = Vec::new();
    for (app_id, rid) in &targets {
        match super::build_app_scoped_storage_config(&manifest, manifest_path, app_id) {
            Ok(storage_config) => {
                if let Err(err) = super::ensure_mutating_storage_access(&storage_config, "compact releases") {
                    preflight_errors.push(format!("{app_id}/{rid}: {err}"));
                } else {
                    prepared_targets.push((app_id.clone(), rid.clone(), storage_config));
                }
            }
            Err(err) => preflight_errors.push(format!("{app_id}/{rid}: {err}")),
        }
    }

    if !preflight_errors.is_empty() {
        return Err(SurgeError::Storage(format!(
            "{} target(s) cannot be compacted: {}",
            preflight_errors.len(),
            preflight_errors.join("; ")
        )));
    }

    let total_targets = prepared_targets.len();
    if let Some(channel) = requested_channel {
        logline::info(&format!("Compacting {total_targets} target(s) on channel '{channel}'"));
    } else {
        logline::info(&format!(
            "Compacting {total_targets} target(s) with auto-discovered channel selection"
        ));
    }
    logline::plain("");

    let mut errors = Vec::new();
    for (app_id, rid, storage_config) in &prepared_targets {
        if let Err(e) = compact_single(app_id, rid, requested_channel, storage_config).await {
            logline::warn(&format!("  Failed {app_id}/{rid}: {e}"));
            errors.push(format!("{app_id}/{rid}: {e}"));
        }
        logline::plain("");
    }

    if errors.is_empty() {
        logline::success(&format!("All {total_targets} target(s) compacted successfully."));
        Ok(())
    } else {
        Err(SurgeError::Storage(format!(
            "{} target(s) failed: {}",
            errors.len(),
            errors.join("; ")
        )))
    }
}

async fn compact_single(
    app_id: &str,
    rid: &str,
    requested_channel: Option<&str>,
    storage_config: &surge_core::context::StorageConfig,
) -> Result<()> {
    const TOTAL_STAGES: usize = 6;

    let theme = UiTheme::global();
    let started = Instant::now();

    print_stage(theme, 1, TOTAL_STAGES, &format!("{app_id}/{rid}"));
    let backend = storage::create_storage_backend(storage_config)?;

    let mut index = match backend.get_object(RELEASES_FILE_COMPRESSED).await {
        Ok(data) => decompress_release_index(&data)?,
        Err(SurgeError::NotFound(_)) => {
            print_stage_done(theme, 1, TOTAL_STAGES, "No release index, skipped");
            return Ok(());
        }
        Err(e) => return Err(e),
    };
    let total_before = index.releases.len();

    print_stage(theme, 2, TOTAL_STAGES, "Resolving channel");
    let Some(resolved_channel) = resolve_compact_channel(app_id, rid, &index, requested_channel)? else {
        print_stage_done(theme, 2, TOTAL_STAGES, "No channels found for target, skipped");
        return Ok(());
    };
    let channel = resolved_channel.name;
    print_stage_done(theme, 2, TOTAL_STAGES, &resolved_channel.note);

    print_stage(
        theme,
        3,
        TOTAL_STAGES,
        &format!("Finding latest release on '{channel}'"),
    );
    let latest_version = index
        .releases
        .iter()
        .filter(|r| r.rid == rid && r.channels.iter().any(|existing| existing == &channel))
        .max_by(|a, b| compare_versions(&a.version, &b.version))
        .map(|r| r.version.clone());

    let Some(latest_version) = latest_version else {
        print_stage_done(theme, 3, TOTAL_STAGES, &format!("No releases on '{channel}', skipped"));
        return Ok(());
    };
    print_stage_done(theme, 3, TOTAL_STAGES, &format!("v{latest_version}"));

    print_stage(theme, 4, TOTAL_STAGES, "Ensuring latest full artifact exists");
    let full_materialized = super::ensure_release_full_artifact(&*backend, &index, rid, &latest_version).await?;
    print_stage_done(
        theme,
        4,
        TOTAL_STAGES,
        if full_materialized {
            "Rebuilt and uploaded latest full artifact"
        } else {
            "Latest full artifact already present"
        },
    );

    print_stage(
        theme,
        5,
        TOTAL_STAGES,
        "Pruning compacted channel history and stale artifacts",
    );
    let stale_filenames = referenced_artifacts(&index);
    let channel_name = channel.as_str();
    let releases_on_channel_before = index
        .releases
        .iter()
        .filter(|release| release.rid == rid && release.channels.iter().any(|existing| existing == channel_name))
        .count();

    for release in &mut index.releases {
        if release.rid != rid || !release.channels.iter().any(|existing| existing == channel_name) {
            continue;
        }

        if release.version == latest_version {
            release.set_primary_delta(None);
        } else {
            release.channels.retain(|existing| existing != channel_name);
        }
    }
    index
        .releases
        .retain(|release| release.rid != rid || !release.channels.is_empty());

    let required = required_artifacts_for_index(&index);

    let mut deleted = 0usize;
    for key in &stale_filenames {
        if required.contains(key) {
            continue;
        }

        match backend.delete_object(key).await {
            Ok(()) | Err(SurgeError::NotFound(_)) => {
                deleted += 1;
            }
            Err(e) => {
                tracing::warn!("Failed to delete {key}: {e}");
            }
        }
    }

    index.last_write_utc = chrono::Utc::now().to_rfc3339();
    let compressed = compress_release_index(&index, DEFAULT_ZSTD_LEVEL)?;
    backend
        .put_object(RELEASES_FILE_COMPRESSED, &compressed, "application/octet-stream")
        .await?;

    let removed = total_before - index.releases.len();
    print_stage_done(
        theme,
        5,
        TOTAL_STAGES,
        &format!(
            "Pruned {older} older '{channel}' release(s), removed {removed} release row(s), deleted {deleted} artifact(s)",
            older = releases_on_channel_before.saturating_sub(1)
        ),
    );

    print_stage_done(
        theme,
        6,
        TOTAL_STAGES,
        &format!(
            "Compacted channel '{channel}' to v{latest_version} (full only) in {}",
            format_duration(started.elapsed())
        ),
    );
    Ok(())
}

struct ResolvedCompactChannel {
    name: String,
    note: String,
}

fn resolve_compact_channel(
    app_id: &str,
    rid: &str,
    index: &surge_core::releases::manifest::ReleaseIndex,
    requested_channel: Option<&str>,
) -> Result<Option<ResolvedCompactChannel>> {
    let available_channels = collect_available_channels_for_rid(index, rid);

    if let Some(channel) = requested_channel {
        if available_channels.is_empty() || available_channels.iter().any(|existing| existing == channel) {
            return Ok(Some(ResolvedCompactChannel {
                name: channel.to_string(),
                note: format!("Using requested channel '{channel}'"),
            }));
        }

        return Err(SurgeError::Config(format!(
            "Channel '{channel}' is not available for target {app_id}/{rid}. Available channels: {}.",
            available_channels.join(", ")
        )));
    }

    match available_channels.as_slice() {
        [] => Ok(None),
        [channel] => Ok(Some(ResolvedCompactChannel {
            name: channel.clone(),
            note: format!("Available channels: {channel}. Selected '{channel}' automatically."),
        })),
        _ => Err(SurgeError::Config(format!(
            "Multiple channels available for target {app_id}/{rid}: {}. Specify --channel <name> to choose.",
            available_channels.join(", ")
        ))),
    }
}

fn collect_available_channels_for_rid(index: &surge_core::releases::manifest::ReleaseIndex, rid: &str) -> Vec<String> {
    let mut channels = BTreeSet::new();
    for release in &index.releases {
        if release.rid != rid {
            continue;
        }
        for channel in &release.channels {
            let trimmed = channel.trim();
            if !trimmed.is_empty() {
                channels.insert(trimmed.to_string());
            }
        }
    }
    channels.into_iter().collect()
}

fn print_stage(theme: UiTheme, stage: usize, total: usize, text: &str) {
    let _ = theme;
    logline::info(&format!("[{stage}/{total}] {text}"));
}

fn print_stage_done(theme: UiTheme, stage: usize, total: usize, text: &str) {
    let _ = theme;
    logline::success(&format!("[{stage}/{total}] {text}"));
}

fn referenced_artifacts(index: &surge_core::releases::manifest::ReleaseIndex) -> BTreeSet<String> {
    let mut keys = BTreeSet::new();
    for release in &index.releases {
        let full = release.full_filename.trim();
        if !full.is_empty() {
            keys.insert(full.to_string());
        }
        for delta in release.all_deltas() {
            let key = delta.filename.trim();
            if !key.is_empty() {
                keys.insert(key.to_string());
            }
        }
    }
    keys
}

#[cfg(test)]
mod tests {
    use super::*;
    use surge_core::crypto::sha256::sha256_hex;
    use surge_core::diff::wrapper::bsdiff_buffers;
    use surge_core::platform::detect::current_rid;
    use surge_core::releases::manifest::{DeltaArtifact, ReleaseEntry, ReleaseIndex};

    fn write_manifest(path: &Path, store_dir: &Path, app_id: &str, rid: &str) {
        let yaml = format!(
            r"schema: 1
storage:
  provider: filesystem
  bucket: {bucket}
apps:
  - id: {app_id}
    name: Compact Test App
    main_exe: demoapp
    targets:
      - rid: {rid}
",
            bucket = store_dir.display()
        );
        std::fs::write(path, yaml).unwrap();
    }

    fn make_release(version: &str, rid: &str, full_bytes: &[u8]) -> ReleaseEntry {
        ReleaseEntry {
            version: version.to_string(),
            channels: vec!["stable".to_string()],
            os: rid.split('-').next().unwrap_or("unknown").to_string(),
            rid: rid.to_string(),
            is_genesis: false,
            full_filename: format!("compact-app-{version}-{rid}-full.tar.zst"),
            full_size: i64::try_from(full_bytes.len()).unwrap(),
            full_sha256: sha256_hex(full_bytes),
            full_compression_level: 0,
            full_zstd_workers: 0,
            deltas: Vec::new(),
            preferred_delta_id: String::new(),
            created_utc: String::new(),
            release_notes: String::new(),
            name: "Compact Test App".to_string(),
            main_exe: "demoapp".to_string(),
            install_directory: "demoapp".to_string(),
            supervisor_id: String::new(),
            icon: String::new(),
            shortcuts: Vec::new(),
            persistent_assets: Vec::new(),
            installers: Vec::new(),
            environment: std::collections::BTreeMap::new(),
        }
    }

    fn write_index(path: &Path, releases: Vec<ReleaseEntry>) {
        let index = ReleaseIndex {
            app_id: "compact-app".to_string(),
            releases,
            ..ReleaseIndex::default()
        };
        let compressed = compress_release_index(&index, DEFAULT_ZSTD_LEVEL).unwrap();
        std::fs::write(path, compressed).unwrap();
    }

    #[tokio::test]
    async fn compact_materializes_latest_delta_only_release_before_pruning_history() {
        let tmp = tempfile::tempdir().unwrap();
        let store_dir = tmp.path().join("store");
        let manifest_path = tmp.path().join("surge.yml");
        let rid = current_rid();
        let app_id = "compact-app";
        std::fs::create_dir_all(&store_dir).unwrap();
        write_manifest(&manifest_path, &store_dir, app_id, &rid);

        let full_v1 = b"full-v1".to_vec();
        let full_v2 = b"full-v2-hello".to_vec();
        let full_v3 = b"full-v3-hello-world".to_vec();
        let delta_v2 = zstd::encode_all(bsdiff_buffers(&full_v1, &full_v2).unwrap().as_slice(), 3).unwrap();
        let delta_v3 = zstd::encode_all(bsdiff_buffers(&full_v2, &full_v3).unwrap().as_slice(), 3).unwrap();

        let mut v1 = make_release("1.0.0", &rid, &full_v1);
        v1.set_primary_delta(None);

        let mut v2 = make_release("1.1.0", &rid, &full_v2);
        let v2_delta_key = format!("{app_id}-1.1.0-{rid}-delta.tar.zst");
        v2.set_primary_delta(Some(DeltaArtifact::bsdiff_zstd(
            "primary",
            "1.0.0",
            &v2_delta_key,
            i64::try_from(delta_v2.len()).unwrap(),
            &sha256_hex(&delta_v2),
        )));

        let mut v3 = make_release("1.2.0", &rid, &full_v3);
        let v3_delta_key = format!("{app_id}-1.2.0-{rid}-delta.tar.zst");
        v3.set_primary_delta(Some(DeltaArtifact::bsdiff_zstd(
            "primary",
            "1.1.0",
            &v3_delta_key,
            i64::try_from(delta_v3.len()).unwrap(),
            &sha256_hex(&delta_v3),
        )));

        std::fs::write(store_dir.join(&v1.full_filename), &full_v1).unwrap();
        std::fs::write(store_dir.join(&v2_delta_key), &delta_v2).unwrap();
        std::fs::write(store_dir.join(&v3_delta_key), &delta_v3).unwrap();
        write_index(
            &store_dir.join(RELEASES_FILE_COMPRESSED),
            vec![v1.clone(), v2, v3.clone()],
        );

        execute(&manifest_path, Some(app_id), Some(&rid), Some("stable"))
            .await
            .unwrap();

        let compacted =
            decompress_release_index(&std::fs::read(store_dir.join(RELEASES_FILE_COMPRESSED)).unwrap()).unwrap();
        assert_eq!(compacted.releases.len(), 1);
        let latest = &compacted.releases[0];
        assert_eq!(latest.version, "1.2.0");
        assert!(latest.selected_delta().is_none());
        assert_eq!(std::fs::read(store_dir.join(&v3.full_filename)).unwrap(), full_v3);
        assert!(!store_dir.join(&v1.full_filename).exists());
        assert!(!store_dir.join(&v2_delta_key).exists());
        assert!(!store_dir.join(&v3_delta_key).exists());
    }

    #[tokio::test]
    async fn compact_prunes_mixed_full_and_delta_history_to_latest_full_only() {
        let tmp = tempfile::tempdir().unwrap();
        let store_dir = tmp.path().join("store");
        let manifest_path = tmp.path().join("surge.yml");
        let rid = current_rid();
        let app_id = "compact-app";
        std::fs::create_dir_all(&store_dir).unwrap();
        write_manifest(&manifest_path, &store_dir, app_id, &rid);

        let full_v1 = b"release-1".to_vec();
        let full_v2 = b"release-2-delta".to_vec();
        let full_v3 = b"release-3-delta".to_vec();
        let full_v4 = b"release-4-full".to_vec();
        let full_v5 = b"release-5-delta".to_vec();
        let full_v6 = b"release-6-full".to_vec();
        let delta_v2 = zstd::encode_all(bsdiff_buffers(&full_v1, &full_v2).unwrap().as_slice(), 3).unwrap();
        let delta_v3 = zstd::encode_all(bsdiff_buffers(&full_v2, &full_v3).unwrap().as_slice(), 3).unwrap();
        let delta_v5 = zstd::encode_all(bsdiff_buffers(&full_v4, &full_v5).unwrap().as_slice(), 3).unwrap();

        let mut v1 = make_release("1.0.0", &rid, &full_v1);
        v1.set_primary_delta(None);

        let mut v2 = make_release("1.1.0", &rid, &full_v2);
        let v2_delta_key = format!("{app_id}-1.1.0-{rid}-delta.tar.zst");
        v2.set_primary_delta(Some(DeltaArtifact::bsdiff_zstd(
            "primary",
            "1.0.0",
            &v2_delta_key,
            i64::try_from(delta_v2.len()).unwrap(),
            &sha256_hex(&delta_v2),
        )));

        let mut v3 = make_release("1.2.0", &rid, &full_v3);
        let v3_delta_key = format!("{app_id}-1.2.0-{rid}-delta.tar.zst");
        v3.set_primary_delta(Some(DeltaArtifact::bsdiff_zstd(
            "primary",
            "1.1.0",
            &v3_delta_key,
            i64::try_from(delta_v3.len()).unwrap(),
            &sha256_hex(&delta_v3),
        )));

        let mut v4 = make_release("1.3.0", &rid, &full_v4);
        v4.set_primary_delta(None);

        let mut v5 = make_release("1.4.0", &rid, &full_v5);
        let v5_delta_key = format!("{app_id}-1.4.0-{rid}-delta.tar.zst");
        v5.set_primary_delta(Some(DeltaArtifact::bsdiff_zstd(
            "primary",
            "1.3.0",
            &v5_delta_key,
            i64::try_from(delta_v5.len()).unwrap(),
            &sha256_hex(&delta_v5),
        )));

        let mut v6 = make_release("1.5.0", &rid, &full_v6);
        v6.set_primary_delta(None);

        std::fs::write(store_dir.join(&v1.full_filename), &full_v1).unwrap();
        std::fs::write(store_dir.join(&v2_delta_key), &delta_v2).unwrap();
        std::fs::write(store_dir.join(&v3_delta_key), &delta_v3).unwrap();
        std::fs::write(store_dir.join(&v4.full_filename), &full_v4).unwrap();
        std::fs::write(store_dir.join(&v5_delta_key), &delta_v5).unwrap();
        std::fs::write(store_dir.join(&v6.full_filename), &full_v6).unwrap();
        write_index(
            &store_dir.join(RELEASES_FILE_COMPRESSED),
            vec![v1.clone(), v2, v3, v4.clone(), v5, v6.clone()],
        );

        execute(&manifest_path, Some(app_id), Some(&rid), Some("stable"))
            .await
            .unwrap();

        let compacted =
            decompress_release_index(&std::fs::read(store_dir.join(RELEASES_FILE_COMPRESSED)).unwrap()).unwrap();
        assert_eq!(compacted.releases.len(), 1);
        let latest = &compacted.releases[0];
        assert_eq!(latest.version, "1.5.0");
        assert!(latest.selected_delta().is_none());
        assert_eq!(std::fs::read(store_dir.join(&v6.full_filename)).unwrap(), full_v6);
        assert!(!store_dir.join(&v1.full_filename).exists());
        assert!(!store_dir.join(&v2_delta_key).exists());
        assert!(!store_dir.join(&v3_delta_key).exists());
        assert!(!store_dir.join(&v4.full_filename).exists());
        assert!(!store_dir.join(&v5_delta_key).exists());
    }

    #[tokio::test]
    async fn compact_auto_selects_single_available_channel_when_channel_omitted() {
        let tmp = tempfile::tempdir().unwrap();
        let store_dir = tmp.path().join("store");
        let manifest_path = tmp.path().join("surge.yml");
        let rid = current_rid();
        let app_id = "compact-app";
        std::fs::create_dir_all(&store_dir).unwrap();
        write_manifest(&manifest_path, &store_dir, app_id, &rid);

        let full_v1 = b"full-v1".to_vec();
        let full_v2 = b"full-v2".to_vec();
        let delta_v2 = zstd::encode_all(bsdiff_buffers(&full_v1, &full_v2).unwrap().as_slice(), 3).unwrap();

        let mut v1 = make_release("1.0.0", &rid, &full_v1);
        v1.channels = vec!["test".to_string()];
        v1.set_primary_delta(None);

        let mut v2 = make_release("1.1.0", &rid, &full_v2);
        v2.channels = vec!["test".to_string()];
        let v2_delta_key = format!("{app_id}-1.1.0-{rid}-delta.tar.zst");
        v2.set_primary_delta(Some(DeltaArtifact::bsdiff_zstd(
            "primary",
            "1.0.0",
            &v2_delta_key,
            i64::try_from(delta_v2.len()).unwrap(),
            &sha256_hex(&delta_v2),
        )));

        std::fs::write(store_dir.join(&v1.full_filename), &full_v1).unwrap();
        std::fs::write(store_dir.join(&v2_delta_key), &delta_v2).unwrap();
        write_index(&store_dir.join(RELEASES_FILE_COMPRESSED), vec![v1, v2.clone()]);

        execute(&manifest_path, Some(app_id), Some(&rid), None).await.unwrap();

        let compacted =
            decompress_release_index(&std::fs::read(store_dir.join(RELEASES_FILE_COMPRESSED)).unwrap()).unwrap();
        assert_eq!(compacted.releases.len(), 1);
        assert_eq!(compacted.releases[0].version, "1.1.0");
        assert_eq!(compacted.releases[0].channels, vec!["test".to_string()]);
        assert!(compacted.releases[0].selected_delta().is_none());
        assert_eq!(std::fs::read(store_dir.join(&v2.full_filename)).unwrap(), full_v2);
    }

    #[tokio::test]
    async fn compact_requires_explicit_channel_when_multiple_channels_exist() {
        let tmp = tempfile::tempdir().unwrap();
        let store_dir = tmp.path().join("store");
        let manifest_path = tmp.path().join("surge.yml");
        let rid = current_rid();
        let app_id = "compact-app";
        std::fs::create_dir_all(&store_dir).unwrap();
        write_manifest(&manifest_path, &store_dir, app_id, &rid);

        let mut stable = make_release("1.0.0", &rid, b"stable-release");
        stable.channels = vec!["stable".to_string()];

        let mut test = make_release("1.1.0", &rid, b"test-release");
        test.channels = vec!["test".to_string()];

        std::fs::write(store_dir.join(&stable.full_filename), b"stable-release").unwrap();
        std::fs::write(store_dir.join(&test.full_filename), b"test-release").unwrap();
        write_index(&store_dir.join(RELEASES_FILE_COMPRESSED), vec![stable, test]);

        let err = execute(&manifest_path, Some(app_id), Some(&rid), None)
            .await
            .unwrap_err();

        let message = err.to_string();
        assert!(message.contains("Multiple channels available"));
        assert!(message.contains("stable, test"));
    }
}
