//! Data export handlers for projects (JSON), transactions (CSV),
//! followers/subscribers (CSV), and content files (ZIP).

mod content;
pub(super) use content::export_content;

use std::fmt::Write as _;

use crate::{
    auth::AuthUser,
    db,
    error::{AppError, Result, ResultExt},
    helpers::{is_htmx_request, sanitize_csv_cell},
    templates::{ExportDownloadTemplate, FormStatusTemplate},
};
use axum::{
    body::Body,
    extract::State,
    http::header::HeaderMap,
    response::{IntoResponse, Response},
};
use bytes::Bytes;
use sqlx::PgPool;
use tokio::sync::mpsc;
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

/// Rows per page for streamed CSV exports.
const EXPORT_BATCH: i64 = 2_000;
/// Hard ceiling on rows in one export, bounding worst-case OFFSET scan cost.
const EXPORT_MAX_ROWS: usize = 1_000_000;

/// Spawn a producer that pages a single export query and streams CSV chunks
/// (header first) into a bounded channel. Peak memory is one batch and the DB
/// connection is released between batches, so a huge export no longer loads the
/// whole result set into one `String` from one unbounded query (ultra-fuzz Run 4
/// S1). `page(limit, offset)` returns the formatted CSV for that page and its row
/// count; a short page ends the stream. OFFSET pagination is sufficient at
/// current scale; keyset is the future optimization (see the `_page` queries).
fn spawn_paginated_csv<F, Fut>(header: &'static str, mut page: F) -> mpsc::Receiver<Bytes>
where
    F: FnMut(i64, i64) -> Fut + Send + 'static,
    Fut: std::future::Future<Output = Result<(String, usize)>> + Send,
{
    let (tx, rx) = mpsc::channel::<Bytes>(4);
    tokio::spawn(async move {
        if tx
            .send(Bytes::from_static(header.as_bytes()))
            .await
            .is_err()
        {
            return;
        }
        let mut offset = 0i64;
        let mut total = 0usize;
        loop {
            let (chunk, n) = match page(EXPORT_BATCH, offset).await {
                Ok(p) => p,
                Err(e) => {
                    tracing::error!(error = ?e, "csv export page failed mid-stream");
                    break;
                }
            };
            if n > 0 && tx.send(Bytes::from(chunk)).await.is_err() {
                return; // client disconnected; stop producing
            }
            offset += n as i64;
            total += n;
            if (n as i64) < EXPORT_BATCH {
                break;
            }
            if total >= EXPORT_MAX_ROWS {
                let _ = tx
                    .send(Bytes::from_static(
                        b"# export truncated at row limit; contact support for a full export\n",
                    ))
                    .await;
                break;
            }
        }
    });
    rx
}

/// Turn a CSV chunk stream into a response. The non-HTMX path (the JS
/// `fetch().then(r => r.blob())` download) gets a true streamed attachment with
/// flat memory. The no-JS HTMX fallback can't stream a `data:` URI, so it
/// materializes a bounded prefix and notes the truncation.
async fn finish_csv(
    is_htmx: bool,
    filename: &str,
    mut rx: mpsc::Receiver<Bytes>,
) -> Result<Response> {
    if is_htmx {
        const HTMX_BYTE_CAP: usize = 4 * 1024 * 1024;
        let mut body = String::new();
        while let Some(chunk) = rx.recv().await {
            body.push_str(&String::from_utf8_lossy(&chunk));
            if body.len() >= HTMX_BYTE_CAP {
                body.push_str(
                    "\n# Export truncated. Enable JavaScript to download the full file.\n",
                );
                break;
            }
        }
        // Dropping `rx` here stops the producer (its next send errors out).
        let data_uri = format!("data:text/csv;charset=utf-8,{}", urlencoding::encode(&body));
        return Ok(ExportDownloadTemplate {
            data_uri,
            filename: filename.to_string(),
        }
        .into_response());
    }
    let stream = ReceiverStream::new(rx).map(Ok::<Bytes, std::convert::Infallible>);
    Response::builder()
        .header("Content-Type", "text/csv")
        .header(
            "Content-Disposition",
            format!("attachment; filename=\"{filename}\""),
        )
        .body(Body::from_stream(stream))
        .context("build streaming export response")
}

/// Return an inline error message for HTMX export requests instead of
/// letting the error propagate to the JSON error layer (which would swap
/// raw JSON text into the status div).
pub(crate) fn export_error_html(message: &str) -> Result<Response> {
    Ok(axum::response::Html(
        FormStatusTemplate {
            success: false,
            message: message.to_string(),
        }
        .render_string()?,
    )
    .into_response())
}

/// HTMX success panel for a queued (backgrounded) export, the link is delivered
/// by email when the job finishes rather than inline.
pub(crate) fn export_pending_html(message: &str) -> Result<Response> {
    Ok(axum::response::Html(
        FormStatusTemplate {
            success: true,
            message: message.to_string(),
        }
        .render_string()?,
    )
    .into_response())
}

/// Build an HTTP response for a downloadable file attachment.
fn download_response(content: Vec<u8>, filename: &str, content_type: &str) -> Result<Response> {
    Response::builder()
        .header("Content-Type", content_type)
        .header(
            "Content-Disposition",
            format!("attachment; filename=\"{filename}\""),
        )
        .body(content.into())
        .context("build download response")
}

// Export API

/// Export all projects and items as a downloadable JSON file.
#[tracing::instrument(skip_all, name = "exports::export_projects")]
pub(super) async fn export_projects(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);

    // Get all projects and all items in 2 queries (not N+1)
    let projects = db::projects::get_projects_by_user(&db, user.id).await?;
    let all_items = db::items::get_items_by_user(&db, user.id).await?;

    // DoS backstop: this handler materializes the entire catalog into an
    // in-memory JSON tree and string on the request path. The ceiling is far
    // above any real creator's catalog, so it never truncates a legitimate
    // "full export". It only refuses a pathological one with a clear message
    // rather than pinning unbounded memory (Run 20 Perf).
    if all_items.len() > EXPORT_MAX_ROWS {
        tracing::warn!(user_id = %user.id, items = all_items.len(), "project export exceeds row ceiling, refusing");
        let msg = "Your catalog is too large to export in a single request. Contact info@makenot.work for a bulk export.";
        if is_htmx {
            return export_error_html(msg);
        }
        return Err(AppError::BadRequest(msg.to_string()));
    }

    let all_item_ids: Vec<db::ItemId> = all_items.iter().map(|i| i.id).collect();
    let tags_map = db::tags::get_tags_for_items(&db, &all_item_ids).await?;

    // Batch-load per-item data (chapters, versions, license keys, item-scoped promo codes)
    let chapters_map = db::chapters::get_chapters_by_items(&db, &all_item_ids).await?;
    let versions_map = db::versions::get_versions_by_items(&db, &all_item_ids).await?;
    let license_keys_map = db::license_keys::get_license_keys_by_items(&db, &all_item_ids).await?;
    let item_promo_codes_map =
        db::promo_codes::get_promo_codes_by_items(&db, &all_item_ids).await?;

    // Promo codes are creator-scoped, fetch once
    let promo_codes = db::promo_codes::get_promo_codes_by_creator(&db, user.id).await?;
    let promo_codes_data: Vec<serde_json::Value> = promo_codes
        .iter()
        .map(|pc| {
            serde_json::json!({
                "code": pc.code,
                "code_purpose": pc.code_purpose.to_string(),
                "discount_type": pc.discount_type.map(|dt| dt.to_string()),
                "discount_value": pc.discount_value,
                "min_price_cents": pc.min_price_cents,
                "trial_days": pc.trial_days,
                "max_uses": pc.max_uses,
                "use_count": pc.use_count,
                "expires_at": pc.expires_at,
                "item_id": pc.item_id,
                "project_id": pc.project_id,
                "tier_id": pc.tier_id,
                "created_at": pc.created_at,
            })
        })
        .collect();

    // Group items by project_id
    let mut items_by_project: std::collections::HashMap<db::ProjectId, Vec<&db::DbItem>> =
        std::collections::HashMap::new();
    for item in &all_items {
        items_by_project
            .entry(item.project_id)
            .or_default()
            .push(item);
    }

    // Batch-load per-project data (blog posts, bundle maps)
    let project_ids: Vec<db::ProjectId> = projects.iter().map(|p| p.id).collect();
    let blog_posts_map = db::blog_posts::get_blog_posts_by_projects(&db, &project_ids).await?;
    let bundle_pairs = db::bundles::get_bundle_maps_by_projects(&db, &project_ids).await?;
    let mut bundle_map: std::collections::HashMap<db::ItemId, Vec<db::ItemId>> =
        std::collections::HashMap::new();
    for (bundle_id, child_id) in &bundle_pairs {
        bundle_map.entry(*bundle_id).or_default().push(*child_id);
    }

    let mut export_data = Vec::new();
    for project in &projects {
        let items = items_by_project
            .get(&project.id)
            .map_or(&[][..], |v| v.as_slice());

        let mut items_data = Vec::new();
        for item in items {
            let tag_names: Vec<&str> = tags_map
                .get(&item.id)
                .map(|tags| tags.iter().map(|t| t.tag_name.as_str()).collect())
                .unwrap_or_default();

            // Content-type-specific fields
            let content_fields = match item.content() {
                db::ContentData::Text {
                    ref body,
                    word_count,
                    reading_time_minutes,
                } => {
                    serde_json::json!({
                        "body": body,
                        "word_count": word_count,
                        "reading_time_minutes": reading_time_minutes,
                    })
                }
                db::ContentData::Audio {
                    duration_seconds,
                    episode_number,
                    ..
                } => {
                    serde_json::json!({
                        "duration_seconds": duration_seconds,
                        "episode_number": episode_number,
                    })
                }
                db::ContentData::Video {
                    duration_seconds,
                    width,
                    height,
                    ..
                } => {
                    serde_json::json!({
                        "duration_seconds": duration_seconds,
                        "width": width,
                        "height": height,
                    })
                }
                db::ContentData::Other => serde_json::json!({}),
            };

            // Chapters (from batch map)
            let chapters_data: Vec<serde_json::Value> = chapters_map
                .get(&item.id)
                .map(|chapters| {
                    chapters
                        .iter()
                        .map(|ch| {
                            serde_json::json!({
                                "title": ch.title,
                                "start_seconds": ch.start_seconds,
                                "sort_order": ch.sort_order,
                            })
                        })
                        .collect()
                })
                .unwrap_or_default();

            // Versions (from batch map)
            let versions_data: Vec<serde_json::Value> = versions_map
                .get(&item.id)
                .map(|versions| {
                    versions
                        .iter()
                        .map(|v| {
                            serde_json::json!({
                                "version_number": v.version_number,
                                "changelog": v.changelog,
                                "file_name": v.file_name,
                                "file_size_bytes": v.file_size_bytes,
                                "is_current": v.is_current,
                                "download_count": v.download_count,
                                "created_at": v.created_at,
                            })
                        })
                        .collect()
                })
                .unwrap_or_default();

            // License keys (from batch map)
            let license_keys_data: Vec<serde_json::Value> = license_keys_map
                .get(&item.id)
                .map(|keys| {
                    keys.iter()
                        .map(|lk| {
                            serde_json::json!({
                                "key_code": lk.key_code,
                                "max_activations": lk.max_activations,
                                "activation_count": lk.activation_count,
                                "revoked_at": lk.revoked_at,
                                "created_at": lk.created_at,
                            })
                        })
                        .collect()
                })
                .unwrap_or_default();

            // Item-scoped promo codes (from batch map)
            let item_promo_codes_data: Vec<serde_json::Value> = item_promo_codes_map
                .get(&item.id)
                .map(|codes| {
                    codes
                        .iter()
                        .map(|pc| {
                            serde_json::json!({
                                "code": pc.code,
                                "code_purpose": pc.code_purpose.to_string(),
                                "max_uses": pc.max_uses,
                                "use_count": pc.use_count,
                                "expires_at": pc.expires_at,
                                "created_at": pc.created_at,
                            })
                        })
                        .collect()
                })
                .unwrap_or_default();

            let mut item_json = serde_json::json!({
                "id": item.id,
                "title": item.title,
                "description": item.description,
                "item_type": item.item_type,
                "price_cents": item.price_cents,
                "is_public": item.is_public,
                "tags": tag_names,
                "play_count": item.play_count,
                "download_count": item.download_count,
                "created_at": item.created_at,
                "chapters": chapters_data,
                "versions": versions_data,
                "license_keys": license_keys_data,
                "promo_codes": item_promo_codes_data,
            });

            // Merge content-type fields into item JSON
            if let Some(obj) = content_fields.as_object() {
                for (k, v) in obj {
                    item_json[k] = v.clone();
                }
            }

            // Include child item IDs for bundles
            if item.item_type == db::ItemType::Bundle
                && let Some(child_ids) = bundle_map.get(&item.id)
            {
                item_json["bundle_items"] = serde_json::json!(child_ids);
            }

            items_data.push(item_json);
        }

        // Blog posts (from batch map)
        let blog_posts_data: Vec<serde_json::Value> = blog_posts_map
            .get(&project.id)
            .map(|posts| {
                posts
                    .iter()
                    .map(|post| {
                        serde_json::json!({
                            "id": post.id,
                            "title": post.title,
                            "slug": post.slug,
                            "body_markdown": post.body_markdown,
                            "published_at": post.published_at,
                            "created_at": post.created_at,
                            "updated_at": post.updated_at,
                        })
                    })
                    .collect()
            })
            .unwrap_or_default();

        export_data.push(serde_json::json!({
            "id": project.id,
            "slug": project.slug,
            "title": project.title,
            "description": project.description,
            "project_type": project.project_type,
            "is_public": project.is_public,
            "created_at": project.created_at,
            "items": items_data,
            "blog_posts": blog_posts_data,
        }));
    }

    // Collections (batch-loaded to avoid N+1)
    let collections = db::collections::get_collections_by_user(&db, user.id).await?;
    let collection_ids: Vec<db::CollectionId> = collections.iter().map(|c| c.id).collect();
    let collection_items_map =
        db::collections::get_item_ids_by_collections(&db, &collection_ids).await?;
    let mut collections_data = Vec::new();
    for c in &collections {
        let item_ids = collection_items_map.get(&c.id).cloned().unwrap_or_default();
        collections_data.push(serde_json::json!({
            "id": c.id,
            "slug": c.slug,
            "title": c.title,
            "description": c.description,
            "is_public": c.is_public,
            "item_ids": item_ids,
            "created_at": c.created_at,
        }));
    }

    let custom_domain = db::custom_domains::get_custom_domain_by_user(&db, user.id).await?;
    let custom_domain_data = custom_domain.map(|d| {
        serde_json::json!({
            "domain": d.domain,
            "verified": d.verified,
        })
    });

    let json_content = serde_json::to_string_pretty(&serde_json::json!({
        "exported_at": chrono::Utc::now().to_rfc3339(),
        "projects": export_data,
        "promo_codes": promo_codes_data,
        "collections": collections_data,
        "custom_domain": custom_domain_data,
    }))
    .map_err(|e| {
        // A full-catalog export that silently degraded to "{}" would look like a
        // successful 200 while delivering nothing, unacceptable for a trust-
        // critical, no-lock-in export. Surface it as a 500 instead.
        crate::error::AppError::Internal(anyhow::anyhow!("failed to serialize project export: {e}"))
    })?;

    if is_htmx {
        let data_uri = format!(
            "data:application/json;charset=utf-8,{}",
            urlencoding::encode(&json_content)
        );
        return Ok(ExportDownloadTemplate {
            data_uri,
            filename: "makenot-work-projects.json".to_string(),
        }
        .into_response());
    }

    download_response(
        json_content.into_bytes(),
        "makenot-work-projects.json",
        "application/json",
    )
}

/// Export all sales transactions as a downloadable CSV file.
#[tracing::instrument(skip_all, name = "exports::export_sales")]
pub(super) async fn export_sales(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);
    let pool = db.clone();
    let uid = user.id;
    let rx = spawn_paginated_csv(
        "Date,Item ID,Item Title,Amount,Status,Buyer Email\n",
        move |limit, offset| {
            let pool = pool.clone();
            async move {
                let rows = db::transactions::get_seller_transactions_for_export_page(
                    &pool, uid, limit, offset,
                )
                .await?;
                let mut buf = String::new();
                for tx in &rows {
                    let item_title = tx.item_title.as_deref().unwrap_or("[Deleted]");
                    let item_id_str = tx
                        .item_id
                        .map_or_else(|| "[Deleted]".to_string(), |id| id.to_string());
                    let buyer_email = tx.buyer_email.as_deref().unwrap_or("");
                    writeln!(
                        buf,
                        "{},{},{},{},{},{}",
                        tx.created_at.format("%Y-%m-%d %H:%M:%S"),
                        item_id_str,
                        sanitize_csv_cell(item_title),
                        crate::formatting::format_dollars_plain(tx.amount_cents),
                        sanitize_csv_cell(&tx.status.to_string()),
                        sanitize_csv_cell(buyer_email)
                    )
                    .unwrap();
                }
                Ok((buf, rows.len()))
            }
        },
    );
    finish_csv(is_htmx, "makenot-work-sales.csv", rx).await
}

/// Export one item's sales as a downloadable CSV file.
///
/// `1ea96868`. This existed as `static/tab-item-sales.js`, which built the file
/// in the browser by scraping the rendered table and quoting each cell with a
/// bare `"`. That neutralised no formula prefix and escaped no embedded quote,
/// and the Buyer column is `guest_email` -- typed by the buyer at guest
/// checkout, so attacker-controlled and landing in a file the creator opens.
/// Every server-side export in this module already ran `sanitize_csv_cell`;
/// that one was the only export in the tree built client-side and inherited
/// none of it.
///
/// # Why it reads the whole set rather than paging
///
/// The item's own sales tab already calls `get_sales_by_item` and renders every
/// row, so holding the same set here adds no exposure the page did not have.
/// Paging it would want a second query, and this one is item-scoped rather than
/// seller-scoped: it is a page of a creator's history, not the history.
///
/// Ownership is the query's, not a separate check: `get_sales_by_item` takes
/// `seller_id` and filters on it, so another creator's item id returns nothing.
#[tracing::instrument(skip_all, name = "exports::export_item_sales")]
pub(super) async fn export_item_sales(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
    axum::extract::Path(item_id): axum::extract::Path<crate::db::ItemId>,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);
    let sales = db::transactions::get_sales_by_item(&db, item_id, user.id).await?;

    let mut body = String::new();
    for tx in &sales {
        let buyer = tx
            .guest_email
            .clone()
            .or_else(|| tx.buyer_id.map(|_| "Registered user".to_string()))
            .unwrap_or_else(|| "Unknown".to_string());
        writeln!(
            body,
            "{},{},{},{}",
            tx.created_at.format("%Y-%m-%d %H:%M"),
            sanitize_csv_cell(&buyer),
            crate::formatting::format_dollars_plain(tx.amount_cents.as_i64()),
            sanitize_csv_cell(&tx.status.to_string()),
        )
        .unwrap();
    }

    // One page, then done. `spawn_paginated_csv` ends when a page comes back
    // shorter than a batch, so the row count is what terminates it and an
    // oversized set still stops on the empty second page.
    let rows = sales.len();
    let mut once = Some(body);
    let rx = spawn_paginated_csv("Date,Buyer,Amount,Status\n", move |_limit, _offset| {
        let page = once.take();
        async move { Ok(page.map_or_else(|| (String::new(), 0), |text| (text, rows))) }
    });

    finish_csv(is_htmx, "makenot-work-item-sales.csv", rx).await
}

/// Export revenue splits as a downloadable CSV file.
#[tracing::instrument(skip_all, name = "exports::export_splits")]
pub(super) async fn export_splits(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);
    let pool = db.clone();
    let uid = user.id;
    let rx = spawn_paginated_csv(
        "Date,Type,Direction,Recipient,Amount,Split %\n",
        move |limit, offset| {
            let pool = pool.clone();
            async move {
                let splits =
                    db::project_members::get_splits_for_export_page(&pool, uid, limit, offset)
                        .await?;
                let mut buf = String::new();
                for split in &splits {
                    let direction = if split.recipient_id == uid {
                        "incoming"
                    } else {
                        "outgoing"
                    };
                    writeln!(
                        buf,
                        "{},{},{},{},{},{}",
                        split.created_at.format("%Y-%m-%d %H:%M:%S"),
                        sanitize_csv_cell(&split.source_type),
                        direction,
                        sanitize_csv_cell(&split.recipient_username),
                        crate::formatting::format_dollars_plain(split.amount_cents),
                        split.split_percent,
                    )
                    .unwrap();
                }
                Ok((buf, splits.len()))
            }
        },
    );
    finish_csv(is_htmx, "makenot-work-splits.csv", rx).await
}

/// Export all purchase transactions as a downloadable CSV file.
#[tracing::instrument(skip_all, name = "exports::export_purchases")]
pub(super) async fn export_purchases(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);
    let pool = db.clone();
    let uid = user.id;
    let rx = spawn_paginated_csv(
        "Date,Item ID,Item Title,Amount,Status\n",
        move |limit, offset| {
            let pool = pool.clone();
            async move {
                let transactions = db::transactions::get_buyer_transactions_for_export_page(
                    &pool, uid, limit, offset,
                )
                .await?;
                // Batch-fetch titles only for this page's transactions missing
                // the denormalized item_title.
                let missing_title_ids: Vec<db::ItemId> = transactions
                    .iter()
                    .filter(|tx| tx.item_title.is_none())
                    .filter_map(|tx| tx.item_id)
                    .collect();
                let title_lookup: std::collections::HashMap<db::ItemId, String> =
                    db::items::get_item_titles_batch(&pool, &missing_title_ids)
                        .await?
                        .into_iter()
                        .collect();

                let mut buf = String::new();
                for tx in &transactions {
                    let item_title = if let Some(title) = &tx.item_title {
                        title.clone()
                    } else if let Some(item_id) = tx.item_id {
                        title_lookup
                            .get(&item_id)
                            .cloned()
                            .unwrap_or_else(|| "[Deleted]".to_string())
                    } else {
                        "[Deleted]".to_string()
                    };
                    let item_id_str = tx
                        .item_id
                        .map_or_else(|| "[Deleted]".to_string(), |id| id.to_string());
                    writeln!(
                        buf,
                        "{},{},{},{},{}",
                        tx.created_at.format("%Y-%m-%d %H:%M:%S"),
                        item_id_str,
                        sanitize_csv_cell(&item_title),
                        crate::formatting::format_dollars_plain(tx.amount_cents),
                        sanitize_csv_cell(&tx.status.to_string())
                    )
                    .unwrap();
                }
                Ok((buf, transactions.len()))
            }
        },
    );
    finish_csv(is_htmx, "makenot-work-purchases.csv", rx).await
}

/// Export followers and subscribers as a downloadable CSV file.
#[tracing::instrument(skip_all, name = "exports::export_followers")]
pub(super) async fn export_followers(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);
    let pool = db.clone();
    let uid = user.id;

    // Two-section CSV (followers, then subscribers); each section pages
    // independently so the whole thing streams in bounded batches (Run 4 S1).
    let (tx, rx) = mpsc::channel::<Bytes>(4);
    tokio::spawn(async move {
        if tx
            .send(Bytes::from_static(
                b"Section,Username,Display Name,Email,Type,Status,Since\n",
            ))
            .await
            .is_err()
        {
            return;
        }

        let mut offset = 0i64;
        let mut total = 0usize;
        loop {
            let rows =
                match db::follows::get_followers_for_export_page(&pool, uid, EXPORT_BATCH, offset)
                    .await
                {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::error!(error = ?e, "followers export page failed");
                        break;
                    }
                };
            if !rows.is_empty() {
                let mut buf = String::new();
                for f in &rows {
                    writeln!(
                        buf,
                        "Follower,{},{},{},{},,{}",
                        sanitize_csv_cell(&f.username),
                        sanitize_csv_cell(f.display_name.as_deref().unwrap_or("")),
                        sanitize_csv_cell(f.email.as_deref().unwrap_or("")),
                        f.target_type,
                        f.created_at.format("%Y-%m-%d %H:%M:%S"),
                    )
                    .unwrap();
                }
                if tx.send(Bytes::from(buf)).await.is_err() {
                    return;
                }
            }
            offset += rows.len() as i64;
            total += rows.len();
            if (rows.len() as i64) < EXPORT_BATCH || total >= EXPORT_MAX_ROWS {
                break;
            }
        }

        let mut offset = 0i64;
        let mut total = 0usize;
        loop {
            let rows = match db::subscriptions::get_project_subscribers_for_export_page(
                &pool,
                uid,
                EXPORT_BATCH,
                offset,
            )
            .await
            {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(error = ?e, "subscribers export page failed");
                    break;
                }
            };
            if !rows.is_empty() {
                let mut buf = String::new();
                for s in &rows {
                    writeln!(
                        buf,
                        "Subscriber,{},{},,{},{},{}",
                        sanitize_csv_cell(&s.username),
                        sanitize_csv_cell(s.display_name.as_deref().unwrap_or("")),
                        sanitize_csv_cell(&s.tier_name),
                        s.status,
                        s.created_at.format("%Y-%m-%d %H:%M:%S"),
                    )
                    .unwrap();
                }
                if tx.send(Bytes::from(buf)).await.is_err() {
                    return;
                }
            }
            offset += rows.len() as i64;
            total += rows.len();
            if (rows.len() as i64) < EXPORT_BATCH || total >= EXPORT_MAX_ROWS {
                break;
            }
        }
    });
    finish_csv(is_htmx, "makenot-work-followers.csv", rx).await
}

/// Export subscriptions as a downloadable CSV file with full detail.
#[tracing::instrument(skip_all, name = "exports::export_subscriptions")]
pub(super) async fn export_subscriptions(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);
    let pool = db.clone();
    let uid = user.id;
    let rx = spawn_paginated_csv(
        "Project,Tier,Price,Username,Status,Period Start,Period End,Canceled At,Created At\n",
        move |limit, offset| {
            let pool = pool.clone();
            async move {
                let subscriptions =
                    db::subscriptions::get_subscriptions_for_export_page(&pool, uid, limit, offset)
                        .await?;
                let fmt_opt = |dt: Option<chrono::DateTime<chrono::Utc>>| -> String {
                    dt.map(|d| d.format("%Y-%m-%d %H:%M:%S").to_string())
                        .unwrap_or_default()
                };
                let mut buf = String::new();
                for s in &subscriptions {
                    writeln!(
                        buf,
                        "{},{},{},{},{},{},{},{},{}",
                        sanitize_csv_cell(&s.project_title),
                        sanitize_csv_cell(&s.tier_name),
                        crate::formatting::format_dollars_plain(s.price_cents),
                        sanitize_csv_cell(&s.username),
                        sanitize_csv_cell(&s.status.to_string()),
                        fmt_opt(s.current_period_start),
                        fmt_opt(s.current_period_end),
                        fmt_opt(s.canceled_at),
                        s.created_at.format("%Y-%m-%d %H:%M:%S"),
                    )
                    .unwrap();
                }
                Ok((buf, subscriptions.len()))
            }
        },
    );
    finish_csv(is_htmx, "makenot-work-subscriptions.csv", rx).await
}

/// Export buyer contacts (who opted to share their email) as CSV.
#[tracing::instrument(skip_all, name = "exports::export_contacts")]
pub(super) async fn export_contacts(
    State(db): State<PgPool>,
    headers: HeaderMap,
    AuthUser(user): AuthUser,
) -> Result<Response> {
    let is_htmx = is_htmx_request(&headers);
    let pool = db.clone();
    let uid = user.id;
    let rx = spawn_paginated_csv(
        "Username,Email,Purchases,Total Spent,Last Purchase\n",
        move |limit, offset| {
            let pool = pool.clone();
            async move {
                let contacts =
                    db::transactions::get_seller_contacts_page(&pool, uid, limit, offset).await?;
                let mut buf = String::new();
                for c in &contacts {
                    writeln!(
                        buf,
                        "{},{},{},{},{}",
                        sanitize_csv_cell(&c.username),
                        sanitize_csv_cell(&c.email),
                        c.total_purchases,
                        crate::formatting::format_dollars_plain(c.total_spent_cents),
                        c.last_purchase_at.format("%Y-%m-%d"),
                    )
                    .unwrap();
                }
                Ok((buf, contacts.len()))
            }
        },
    );
    finish_csv(is_htmx, "makenot-work-contacts.csv", rx).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::to_bytes;
    use axum::http::StatusCode;

    #[test]
    fn download_response_sets_content_type() {
        let resp = download_response(b"hello".to_vec(), "test.csv", "text/csv").unwrap();
        assert_eq!(resp.headers().get("Content-Type").unwrap(), "text/csv");
    }

    #[test]
    fn download_response_sets_content_disposition() {
        let resp = download_response(b"data".to_vec(), "export.json", "application/json").unwrap();
        let disp = resp
            .headers()
            .get("Content-Disposition")
            .unwrap()
            .to_str()
            .unwrap();
        assert_eq!(disp, "attachment; filename=\"export.json\"");
    }

    #[test]
    fn download_response_status_200() {
        let resp = download_response(vec![], "empty.csv", "text/csv").unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn download_response_body_matches() {
        let content = b"col1,col2\na,b\n".to_vec();
        let resp = download_response(content.clone(), "f.csv", "text/csv").unwrap();
        let body = to_bytes(resp.into_body(), 1024).await.unwrap();
        assert_eq!(body.as_ref(), content.as_slice());
    }

    #[test]
    fn download_response_filename_with_spaces() {
        let resp = download_response(b"x".to_vec(), "my export.csv", "text/csv").unwrap();
        let disp = resp
            .headers()
            .get("Content-Disposition")
            .unwrap()
            .to_str()
            .unwrap();
        assert!(disp.contains("my export.csv"));
    }
}