//! Content file export: ZIP archive of audio, covers, videos, versions, and insertions.
//!
//! Writes the ZIP to a temporary file and uploads via S3 multipart upload,
//! so peak memory is O(single_file) regardless of total export size.

use std::fmt::Write as _;
use std::io::Write;

use axum::{
    extract::{Query, State},
    http::header::HeaderMap,
    response::Response,
};
use serde::Deserialize;
use zip::write::SimpleFileOptions;

use crate::AppStorage;
use crate::background::BackgroundTx;
use crate::email::EmailClient;
use sqlx::PgPool;

use crate::{
    auth::AuthUser,
    db,
    error::{AppError, Result, ResultExt},
    helpers::is_htmx_request,
};

use super::export_error_html;

/// Max content exports running at once. Each can move up to 2 GB through a
/// synchronous zip on the blocking pool; without a cap a burst could saturate
/// the blocking pool and stall unrelated `spawn_blocking` work. Excess exports
/// queue on the semaphore instead.
const MAX_CONCURRENT_EXPORTS: usize = 3;
static EXPORT_LIMITER: tokio::sync::Semaphore =
    tokio::sync::Semaphore::const_new(MAX_CONCURRENT_EXPORTS);

/// Query parameters for the content export endpoint.
#[derive(Deserialize)]
pub(in crate::routes::api) struct ContentExportQuery {
    /// When set, only export files from this project (useful for large
    /// catalogs or to stay within the 2GB per-export memory limit).
    pub project_id: Option<db::ProjectId>,
}

/// Export content files as a ZIP archive uploaded to S3.
///
/// Collects audio, covers, version downloads, and insertion clips,
/// bundles them with a README.txt manifest, uploads to S3 as a
/// temporary export, and returns a presigned download link.
///
/// Pass `?project_id=<uuid>` to limit the export to a single project
/// (insertions are user-scoped and always excluded from per-project exports).
#[tracing::instrument(skip_all, name = "exports::export_content")]
pub(in crate::routes::api) async fn export_content(
    State(db): State<PgPool>,
    State(storage): State<AppStorage>,
    State(email): State<EmailClient>,
    State(bg): State<BackgroundTx>,
    headers: HeaderMap,
    Query(query): Query<ContentExportQuery>,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);

    let s3 = storage.s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("File storage is not configured".to_string())
    })?;

    // Collect all S3 keys from items, versions, and insertions. These are fast
    // indexed reads, done up front so an empty export is reported immediately;
    // the heavy download + zip + multipart upload runs in the background (below).
    let item_keys = db::items::get_user_s3_keys(&db, user.id).await?;
    let version_keys = db::versions::get_user_version_s3_keys(&db, user.id).await?;

    // Build the list of (s3_key, zip_path, db_size) triples. The DB-known file
    // size lets us enforce the per-file/total caps without a per-file S3 HEAD
    // round-trip (these columns are written at upload-confirm time). `None` only
    // for legacy rows missing the size; those fall through to the post-download
    // total guard.
    let mut files: Vec<(String, String, Option<i64>)> = Vec::new();

    for item in &item_keys {
        if let Some(pid) = query.project_id
            && item.project_id != pid
        {
            continue;
        }
        let slug = item.project_slug.as_str();
        let title = sanitize_filename(&item.title);
        if let Some(ref key) = item.audio_s3_key {
            let ext = extension_from_key(key);
            files.push((
                key.clone(),
                format!("projects/{slug}/{title}.{ext}"),
                item.audio_file_size_bytes,
            ));
        }
        if let Some(ref key) = item.cover_s3_key {
            let ext = extension_from_key(key);
            files.push((
                key.clone(),
                format!("projects/{slug}/{title}-cover.{ext}"),
                item.cover_file_size_bytes,
            ));
        }
        if let Some(ref key) = item.video_s3_key {
            let ext = extension_from_key(key);
            files.push((
                key.clone(),
                format!("projects/{slug}/{title}-video.{ext}"),
                item.video_file_size_bytes,
            ));
        }
    }

    for ver in &version_keys {
        if let Some(pid) = query.project_id
            && ver.project_id != pid
        {
            continue;
        }
        if let Some(ref key) = ver.s3_key {
            let slug = ver.project_slug.as_str();
            let title = sanitize_filename(&ver.item_title);
            let fname = ver.file_name.as_deref().unwrap_or("file");
            files.push((
                key.clone(),
                format!(
                    "projects/{}/{}/v{}-{}",
                    slug, title, ver.version_number, fname
                ),
                ver.file_size_bytes,
            ));
        }
    }

    // Insertions are user-scoped (not project-scoped), so only include
    // them when exporting all content (no project_id filter).
    if query.project_id.is_none() {
        let insertions = db::content_insertions::list_insertions(&db, user.id).await?;
        for ins in &insertions {
            let ext = extension_from_key(&ins.storage_key);
            let title = sanitize_filename(&ins.title);
            files.push((
                ins.storage_key.clone(),
                format!("insertions/{title}.{ext}"),
                Some(ins.file_size),
            ));
        }
    }

    // Exclude confirmed-malicious objects. The download paths refuse to serve a
    // Quarantined object even to its own creator (downloads.rs); the export must
    // honor the same invariant, since it collects raw S3 keys with no scan gate
    // of its own (Run 20 Security). Filtering here (not in get_user_s3_keys)
    // keeps that shared query intact for storage accounting, which must still see
    // every key.
    let export_keys: Vec<String> = files.iter().map(|(k, _, _)| k.clone()).collect();
    let quarantined = db::scanning::quarantined_s3_keys(&db, &export_keys).await?;
    if !quarantined.is_empty() {
        tracing::warn!(
            user_id = %user.id, dropped = quarantined.len(),
            "excluding quarantined objects from content export"
        );
        files.retain(|(key, _, _)| !quarantined.contains(key));
    }

    if files.is_empty() {
        if is_htmx {
            return export_error_html("No content files to export.");
        }
        return Err(AppError::BadRequest(
            "No content files to export.".to_string(),
        ));
    }

    // Hand the heavy work to the background pool: download every file, build the
    // ZIP on disk, multipart-upload it, and email a presigned link. The request
    // returns now instead of holding a connection + export permit for the whole
    // multi-GB job, a wedged S3 GET used to pin both indefinitely.
    let s3 = s3.clone();
    let email = email.clone();
    let username = user.username.to_string();
    let to_email = user.email.clone();
    let to_name = user.display_name.clone();
    let user_id = user.id;
    bg.spawn("content_export", async move {
        match build_content_export(&s3, user_id, &username, files).await {
            Ok(download_url) => {
                if let Err(e) = email
                    .send_content_export_ready(&to_email, to_name.as_deref(), &download_url)
                    .await
                {
                    tracing::error!(error = ?e, "failed to send content-export-ready email");
                }
            }
            Err(reason) => {
                tracing::warn!(user_id = %user_id, reason = %reason, "content export failed");
                if let Err(e) = email
                    .send_content_export_failed(&to_email, to_name.as_deref(), Some(&reason))
                    .await
                {
                    tracing::error!(error = ?e, "failed to send content-export-failed email");
                }
            }
        }
    });

    let message = format!(
        "Preparing your content export. We'll email a download link to {} when it's ready.",
        user.email
    );
    if is_htmx {
        return super::export_pending_html(&message);
    }
    Response::builder()
        .status(axum::http::StatusCode::ACCEPTED)
        .body(message.into())
        .context("build export accepted response")
}

/// Failure modes of [`spool_s3_object_to_file`].
enum SpoolError {
    /// The object exceeded the per-file cap mid-stream (never fully buffered).
    TooLarge,
    /// An S3 or filesystem error; the message is log-only (never user-facing).
    Io(String),
}

/// Stream an S3 object to `dest`, returning the number of bytes written. Aborts
/// with [`SpoolError::TooLarge`] the moment the running total exceeds
/// `max_bytes`, so a mis-sized object can't fill the disk. Peak memory is one
/// streaming chunk, this is what keeps the export's footprint at O(chunk)
/// rather than O(file).
async fn spool_s3_object_to_file(
    s3: &std::sync::Arc<dyn crate::storage::StorageBackend>,
    s3_key: &str,
    dest: &std::path::Path,
    max_bytes: u64,
) -> std::result::Result<u64, SpoolError> {
    use tokio::io::AsyncWriteExt;

    let mut stream = s3
        .download_stream(s3_key)
        .await
        .map_err(|e| SpoolError::Io(e.to_string()))?;
    let mut file = tokio::fs::File::create(dest)
        .await
        .map_err(|e| SpoolError::Io(e.to_string()))?;
    let mut written: u64 = 0;
    while let Some(chunk) = stream
        .try_next()
        .await
        .map_err(|e| SpoolError::Io(e.to_string()))?
    {
        written += chunk.len() as u64;
        if written > max_bytes {
            return Err(SpoolError::TooLarge);
        }
        file.write_all(&chunk)
            .await
            .map_err(|e| SpoolError::Io(e.to_string()))?;
    }
    file.flush()
        .await
        .map_err(|e| SpoolError::Io(e.to_string()))?;
    Ok(written)
}

/// Build the content-export ZIP off the request path: download every file,
/// zip to a tempfile, multipart-upload, and return a 1-hour presigned download
/// URL. Holds the [`EXPORT_LIMITER`] permit for its lifetime so a burst can't
/// saturate the blocking pool. On any failure returns a user-facing reason
/// string (emailed to the creator); never panics the background task.
async fn build_content_export(
    s3: &std::sync::Arc<dyn crate::storage::StorageBackend>,
    user_id: db::UserId,
    username: &str,
    files: Vec<(String, String, Option<i64>)>,
) -> std::result::Result<String, String> {
    // Hold a concurrency permit for the lifetime of the export so a burst can't
    // saturate the blocking pool. Acquired here (off the request path), so a
    // queued export holds no DB connection while it waits.
    let _export_permit = EXPORT_LIMITER
        .acquire()
        .await
        .expect("export limiter semaphore is never closed");

    let s3_clone = s3.clone();

    let tmp_dir = tempfile::tempdir().map_err(|e| format!("create temp dir for export: {e}"))?;
    let zip_path = tmp_dir.path().join("export.zip");

    {
        // The `zip` crate's IO is synchronous; a single `write_all` of up to
        // 500 MB (compression is Stored, so this is raw disk IO) would stall a
        // tokio worker. Every blocking zip operation below runs on the blocking
        // pool via `spawn_blocking`; the writer is moved in and handed back out
        // each step. S3 downloads stay async, and peak memory is still
        // O(largest_single_file), one file is in RAM at a time.
        let create_path = zip_path.clone();
        let mut zip =
            tokio::task::spawn_blocking(move || -> std::result::Result<_, std::io::Error> {
                let zip_file = std::fs::File::create(&create_path)?;
                Ok(zip::ZipWriter::new(std::io::BufWriter::new(zip_file)))
            })
            .await
            .map_err(|e| format!("join zip create task: {e}"))?
            .map_err(|e| format!("create export zip file: {e}"))?;

        let options =
            SimpleFileOptions::default().compression_method(zip::CompressionMethod::Stored);

        let mut manifest: Vec<(String, i64)> = Vec::new();
        let mut total_size: u64 = 0;
        const MAX_TOTAL_SIZE: u64 = 2 * 1024 * 1024 * 1024; // 2 GB
        const MAX_FILE_SIZE: u64 = 500 * 1024 * 1024; // 500 MB per file
        let mut skipped: Vec<String> = Vec::new();

        for (s3_key, zip_path_entry, db_size) in &files {
            // Resolve the file size BEFORE downloading so a single huge object
            // can't blow the heap before any guard fires. Prefer the DB column
            // written at upload-confirm; fall back to a cheap S3 HEAD for legacy
            // rows that predate it (a `None` size). A row whose size can't be
            // resolved (HEAD error / object gone) is skipped rather than blindly
            // buffered, without this, a legacy `None`-size object of arbitrary
            // size landed fully in RAM before the post-download total check could
            // fire (PERF-1).
            let size = match db_size {
                Some(s) => (*s).max(0) as u64,
                None => match s3_clone.object_size(s3_key).await {
                    Ok(Some(s)) => s.max(0) as u64,
                    Ok(None) => {
                        tracing::warn!(s3_key = %s3_key, "export: object has no size (missing?), skipping");
                        skipped.push(zip_path_entry.clone());
                        continue;
                    }
                    Err(e) => {
                        tracing::warn!(s3_key = %s3_key, error = %e, "export: HEAD failed, skipping");
                        skipped.push(zip_path_entry.clone());
                        continue;
                    }
                },
            };
            if size > MAX_FILE_SIZE {
                skipped.push(format!(
                    "{zip_path_entry} (exceeds 500 MB per-file export cap)"
                ));
                continue;
            }
            if total_size + size > MAX_TOTAL_SIZE {
                return Err("Content export exceeds the 2 GB limit. Try exporting a single project instead.".to_string());
            }

            // Stream the object to a per-file spool on disk instead of buffering
            // the whole (up to 500 MB) file in RAM. Peak memory is one streaming
            // chunk, not the full object, three concurrent exports previously
            // held ~1.5 GB of file bytes between them. The scan pipeline streams
            // large files the same way; the exporter now matches it (Run #5 Perf
            // S2). The MAX_FILE_SIZE guard is re-enforced mid-stream as a backstop
            // for legacy rows whose size couldn't be resolved above.
            let part_path = tmp_dir.path().join(format!("part-{}", manifest.len()));
            let part_bytes =
                match spool_s3_object_to_file(&s3_clone, s3_key, &part_path, MAX_FILE_SIZE).await {
                    Ok(n) => n,
                    Err(SpoolError::TooLarge) => {
                        skipped.push(format!(
                            "{zip_path_entry} (exceeds 500 MB per-file export cap)"
                        ));
                        let _ = tokio::fs::remove_file(&part_path).await;
                        continue;
                    }
                    Err(SpoolError::Io(e)) => {
                        tracing::warn!("Failed to download S3 key {}: {}", s3_key, e);
                        skipped.push(zip_path_entry.clone());
                        let _ = tokio::fs::remove_file(&part_path).await;
                        continue;
                    }
                };
            total_size += part_bytes;
            if total_size > MAX_TOTAL_SIZE {
                return Err("Content export exceeds the 2 GB limit. Try exporting a single project instead.".to_string());
            }
            let file_size = part_bytes as i64;
            // Copy the spooled file into the zip on the blocking pool (the zip
            // crate's IO is synchronous). The reader streams from disk via
            // io::copy, so RAM stays at the copy buffer size, not the file size.
            let entry = zip_path_entry.clone();
            let copy_path = part_path.clone();
            zip = tokio::task::spawn_blocking(
                move || -> std::result::Result<_, zip::result::ZipError> {
                    zip.start_file(&entry, options)?;
                    let mut reader = std::fs::File::open(&copy_path)?;
                    std::io::copy(&mut reader, &mut zip)?;
                    Ok(zip)
                },
            )
            .await
            .map_err(|e| format!("join zip write task: {e}"))?
            .map_err(|e| format!("write file into export zip: {e}"))?;
            let _ = tokio::fs::remove_file(&part_path).await;
            manifest.push((zip_path_entry.clone(), file_size));
        }

        if manifest.is_empty() {
            return Err(
                "Could not download any files from storage. Please try again later.".to_string(),
            );
        }

        // Build README.txt as the last ZIP entry (cheap string work, async side)
        let now = chrono::Utc::now();
        let mut readme = format!(
            "Makenotwork Content Export\n\
             Creator: {}\n\
             Exported: {}\n\
             Files: {}\n\n\
             Manifest:\n",
            username,
            now.format("%Y-%m-%d %H:%M:%S UTC"),
            manifest.len(),
        );
        for (path, size) in &manifest {
            writeln!(
                readme,
                "  {path} ({})",
                crate::helpers::format_file_size(*size)
            )
            .unwrap();
        }
        if !skipped.is_empty() {
            writeln!(
                readme,
                "\nSkipped ({} files could not be downloaded):",
                skipped.len()
            )
            .unwrap();
            for path in &skipped {
                writeln!(readme, "  {path}").unwrap();
            }
        }
        readme.push_str("\nNote: Git repositories are not included in this export.\n");
        readme.push_str(
            "Clone them separately: git clone https://makenot.work/git/<username>/<repo>.git\n",
        );

        // Append README, finalize the central directory, and flush the buffer to
        // disk, all blocking, off the runtime before the upload reads the file.
        tokio::task::spawn_blocking(move || -> std::result::Result<(), zip::result::ZipError> {
            zip.start_file("README.txt", options)?;
            zip.write_all(readme.as_bytes())?;
            let buf = zip.finish()?;
            // Flush BufWriter so all bytes hit the OS file before we upload it.
            buf.into_inner()
                .map_err(std::io::IntoInnerError::into_error)?;
            Ok(())
        })
        .await
        .map_err(|e| format!("join zip finalize task: {e}"))?
        .map_err(|e| format!("finalize export zip: {e}"))?;
    }

    // Upload ZIP to S3 via multipart upload (streams from disk in 10 MB parts)
    let timestamp = chrono::Utc::now().format("%Y%m%d-%H%M%S");
    let export_key =
        crate::storage::S3Client::generate_content_export_key(user_id, &timestamp.to_string());
    if let Err(e) = s3
        .upload_multipart(&export_key, "application/zip", &zip_path)
        .await
    {
        tracing::error!(error = ?e, "Failed to upload content export ZIP to S3");
        return Err("Failed to upload the export to storage.".to_string());
    }

    // Generate presigned download URL (1 hour)
    let download_url = s3
        .presign_download(&export_key, Some(3600))
        .await
        .map_err(|e| {
            tracing::error!(error = ?e, "Failed to generate presigned URL for content export");
            "Export created but the download link could not be generated.".to_string()
        })?;

    Ok(download_url)
}

/// Extract file extension from an S3 key (e.g. "user/item/audio/track.mp3" -> "mp3").
fn extension_from_key(key: &str) -> &str {
    // Extract from the basename only: a dot earlier in the path (e.g. a user
    // handle like `alice.dev/...`) must not be mistaken for the extension, and an
    // extensionless key must fall back to "bin" rather than returning the whole
    // path as a bogus extension (audit Run 13).
    let basename = key.rsplit('/').next().unwrap_or(key);
    match basename.rsplit_once('.') {
        Some((stem, ext)) if !stem.is_empty() && !ext.is_empty() => ext,
        _ => "bin",
    }
}

/// Sanitize a title for use as a filename in the ZIP archive.
fn sanitize_filename(name: &str) -> String {
    name.chars()
        .map(|c| {
            if c.is_alphanumeric() || c == '-' || c == '_' || c == ' ' {
                c
            } else {
                '_'
            }
        })
        .collect::<String>()
        .trim()
        .to_string()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn extension_from_key_mp3() {
        assert_eq!(extension_from_key("user/item/audio/track.mp3"), "mp3");
    }

    #[test]
    fn extension_from_key_nested_path() {
        assert_eq!(extension_from_key("a/b/c/file.tar.gz"), "gz");
    }

    #[test]
    fn extension_from_key_no_dot_returns_bin() {
        // An extensionless key falls back to "bin", never the whole path.
        assert_eq!(extension_from_key("user/item/audio/noext"), "bin");
    }

    #[test]
    fn extension_from_key_dot_in_path_not_basename() {
        // A dot earlier in the path is not the extension.
        assert_eq!(extension_from_key("alice.dev/item/noext"), "bin");
        assert_eq!(extension_from_key("alice.dev/item/track.wav"), "wav");
    }

    #[test]
    fn extension_from_key_empty_returns_bin() {
        assert_eq!(extension_from_key(""), "bin");
    }

    #[test]
    fn extension_from_key_dot_only() {
        // Trailing dot → empty ext → "bin".
        assert_eq!(extension_from_key("file."), "bin");
    }

    #[test]
    fn sanitize_filename_passthrough() {
        assert_eq!(sanitize_filename("My Track"), "My Track");
    }

    #[test]
    fn sanitize_filename_special_chars() {
        assert_eq!(sanitize_filename("hello/world:2"), "hello_world_2");
    }

    #[test]
    fn sanitize_filename_preserves_hyphens_underscores() {
        assert_eq!(sanitize_filename("my-file_name"), "my-file_name");
    }

    #[test]
    fn sanitize_filename_trims_whitespace() {
        assert_eq!(sanitize_filename("  padded  "), "padded");
    }

    #[test]
    fn sanitize_filename_empty() {
        assert_eq!(sanitize_filename(""), "");
    }

    #[test]
    fn sanitize_filename_all_special() {
        assert_eq!(sanitize_filename("@#$%"), "____");
    }
}