Skip to main content

max / makenotwork

33.2 KB · 895 lines History Blame Raw
1 //! Data export handlers for projects (JSON), transactions (CSV),
2 //! followers/subscribers (CSV), and content files (ZIP).
3
4 mod content;
5 pub(super) use content::export_content;
6
7 use std::fmt::Write as _;
8
9 use crate::{
10 auth::AuthUser,
11 db,
12 error::{AppError, Result, ResultExt},
13 helpers::{is_htmx_request, sanitize_csv_cell},
14 templates::{ExportDownloadTemplate, FormStatusTemplate},
15 };
16 use axum::{
17 body::Body,
18 extract::State,
19 http::header::HeaderMap,
20 response::{IntoResponse, Response},
21 };
22 use bytes::Bytes;
23 use sqlx::PgPool;
24 use tokio::sync::mpsc;
25 use tokio_stream::{StreamExt, wrappers::ReceiverStream};
26
27 /// Rows per page for streamed CSV exports.
28 const EXPORT_BATCH: i64 = 2_000;
29 /// Hard ceiling on rows in one export, bounding worst-case OFFSET scan cost.
30 const EXPORT_MAX_ROWS: usize = 1_000_000;
31
32 /// Spawn a producer that pages a single export query and streams CSV chunks
33 /// (header first) into a bounded channel. Peak memory is one batch and the DB
34 /// connection is released between batches, so a huge export no longer loads the
35 /// whole result set into one `String` from one unbounded query (ultra-fuzz Run 4
36 /// S1). `page(limit, offset)` returns the formatted CSV for that page and its row
37 /// count; a short page ends the stream. OFFSET pagination is sufficient at
38 /// current scale; keyset is the future optimization (see the `_page` queries).
39 fn spawn_paginated_csv<F, Fut>(header: &'static str, mut page: F) -> mpsc::Receiver<Bytes>
40 where
41 F: FnMut(i64, i64) -> Fut + Send + 'static,
42 Fut: std::future::Future<Output = Result<(String, usize)>> + Send,
43 {
44 let (tx, rx) = mpsc::channel::<Bytes>(4);
45 tokio::spawn(async move {
46 if tx
47 .send(Bytes::from_static(header.as_bytes()))
48 .await
49 .is_err()
50 {
51 return;
52 }
53 let mut offset = 0i64;
54 let mut total = 0usize;
55 loop {
56 let (chunk, n) = match page(EXPORT_BATCH, offset).await {
57 Ok(p) => p,
58 Err(e) => {
59 tracing::error!(error = ?e, "csv export page failed mid-stream");
60 break;
61 }
62 };
63 if n > 0 && tx.send(Bytes::from(chunk)).await.is_err() {
64 return; // client disconnected; stop producing
65 }
66 offset += n as i64;
67 total += n;
68 if (n as i64) < EXPORT_BATCH {
69 break;
70 }
71 if total >= EXPORT_MAX_ROWS {
72 let _ = tx
73 .send(Bytes::from_static(
74 b"# export truncated at row limit; contact support for a full export\n",
75 ))
76 .await;
77 break;
78 }
79 }
80 });
81 rx
82 }
83
84 /// Turn a CSV chunk stream into a response. The non-HTMX path (the JS
85 /// `fetch().then(r => r.blob())` download) gets a true streamed attachment with
86 /// flat memory. The no-JS HTMX fallback can't stream a `data:` URI, so it
87 /// materializes a bounded prefix and notes the truncation.
88 async fn finish_csv(
89 is_htmx: bool,
90 filename: &str,
91 mut rx: mpsc::Receiver<Bytes>,
92 ) -> Result<Response> {
93 if is_htmx {
94 const HTMX_BYTE_CAP: usize = 4 * 1024 * 1024;
95 let mut body = String::new();
96 while let Some(chunk) = rx.recv().await {
97 body.push_str(&String::from_utf8_lossy(&chunk));
98 if body.len() >= HTMX_BYTE_CAP {
99 body.push_str(
100 "\n# Export truncated. Enable JavaScript to download the full file.\n",
101 );
102 break;
103 }
104 }
105 // Dropping `rx` here stops the producer (its next send errors out).
106 let data_uri = format!("data:text/csv;charset=utf-8,{}", urlencoding::encode(&body));
107 return Ok(ExportDownloadTemplate {
108 data_uri,
109 filename: filename.to_string(),
110 }
111 .into_response());
112 }
113 let stream = ReceiverStream::new(rx).map(Ok::<Bytes, std::convert::Infallible>);
114 Response::builder()
115 .header("Content-Type", "text/csv")
116 .header(
117 "Content-Disposition",
118 format!("attachment; filename=\"{filename}\""),
119 )
120 .body(Body::from_stream(stream))
121 .context("build streaming export response")
122 }
123
124 /// Return an inline error message for HTMX export requests instead of
125 /// letting the error propagate to the JSON error layer (which would swap
126 /// raw JSON text into the status div).
127 pub(crate) fn export_error_html(message: &str) -> Result<Response> {
128 Ok(axum::response::Html(
129 FormStatusTemplate {
130 success: false,
131 message: message.to_string(),
132 }
133 .render_string()?,
134 )
135 .into_response())
136 }
137
138 /// HTMX success panel for a queued (backgrounded) export, the link is delivered
139 /// by email when the job finishes rather than inline.
140 pub(crate) fn export_pending_html(message: &str) -> Result<Response> {
141 Ok(axum::response::Html(
142 FormStatusTemplate {
143 success: true,
144 message: message.to_string(),
145 }
146 .render_string()?,
147 )
148 .into_response())
149 }
150
151 /// Build an HTTP response for a downloadable file attachment.
152 fn download_response(content: Vec<u8>, filename: &str, content_type: &str) -> Result<Response> {
153 Response::builder()
154 .header("Content-Type", content_type)
155 .header(
156 "Content-Disposition",
157 format!("attachment; filename=\"{filename}\""),
158 )
159 .body(content.into())
160 .context("build download response")
161 }
162
163 // Export API
164
165 /// Export all projects and items as a downloadable JSON file.
166 #[tracing::instrument(skip_all, name = "exports::export_projects")]
167 pub(super) async fn export_projects(
168 State(db): State<PgPool>,
169 headers: HeaderMap,
170 AuthUser(user): AuthUser,
171 ) -> Result<Response> {
172 let is_htmx = is_htmx_request(&headers);
173
174 // Get all projects and all items in 2 queries (not N+1)
175 let projects = db::projects::get_projects_by_user(&db, user.id).await?;
176 let all_items = db::items::get_items_by_user(&db, user.id).await?;
177
178 // DoS backstop: this handler materializes the entire catalog into an
179 // in-memory JSON tree and string on the request path. The ceiling is far
180 // above any real creator's catalog, so it never truncates a legitimate
181 // "full export". It only refuses a pathological one with a clear message
182 // rather than pinning unbounded memory (Run 20 Perf).
183 if all_items.len() > EXPORT_MAX_ROWS {
184 tracing::warn!(user_id = %user.id, items = all_items.len(), "project export exceeds row ceiling, refusing");
185 let msg = "Your catalog is too large to export in a single request. Contact info@makenot.work for a bulk export.";
186 if is_htmx {
187 return export_error_html(msg);
188 }
189 return Err(AppError::BadRequest(msg.to_string()));
190 }
191
192 let all_item_ids: Vec<db::ItemId> = all_items.iter().map(|i| i.id).collect();
193 let tags_map = db::tags::get_tags_for_items(&db, &all_item_ids).await?;
194
195 // Batch-load per-item data (chapters, versions, license keys, item-scoped promo codes)
196 let chapters_map = db::chapters::get_chapters_by_items(&db, &all_item_ids).await?;
197 let versions_map = db::versions::get_versions_by_items(&db, &all_item_ids).await?;
198 let license_keys_map = db::license_keys::get_license_keys_by_items(&db, &all_item_ids).await?;
199 let item_promo_codes_map =
200 db::promo_codes::get_promo_codes_by_items(&db, &all_item_ids).await?;
201
202 // Promo codes are creator-scoped, fetch once
203 let promo_codes = db::promo_codes::get_promo_codes_by_creator(&db, user.id).await?;
204 let promo_codes_data: Vec<serde_json::Value> = promo_codes
205 .iter()
206 .map(|pc| {
207 serde_json::json!({
208 "code": pc.code,
209 "code_purpose": pc.code_purpose.to_string(),
210 "discount_type": pc.discount_type.map(|dt| dt.to_string()),
211 "discount_value": pc.discount_value,
212 "min_price_cents": pc.min_price_cents,
213 "trial_days": pc.trial_days,
214 "max_uses": pc.max_uses,
215 "use_count": pc.use_count,
216 "expires_at": pc.expires_at,
217 "item_id": pc.item_id,
218 "project_id": pc.project_id,
219 "tier_id": pc.tier_id,
220 "created_at": pc.created_at,
221 })
222 })
223 .collect();
224
225 // Group items by project_id
226 let mut items_by_project: std::collections::HashMap<db::ProjectId, Vec<&db::DbItem>> =
227 std::collections::HashMap::new();
228 for item in &all_items {
229 items_by_project
230 .entry(item.project_id)
231 .or_default()
232 .push(item);
233 }
234
235 // Batch-load per-project data (blog posts, bundle maps)
236 let project_ids: Vec<db::ProjectId> = projects.iter().map(|p| p.id).collect();
237 let blog_posts_map = db::blog_posts::get_blog_posts_by_projects(&db, &project_ids).await?;
238 let bundle_pairs = db::bundles::get_bundle_maps_by_projects(&db, &project_ids).await?;
239 let mut bundle_map: std::collections::HashMap<db::ItemId, Vec<db::ItemId>> =
240 std::collections::HashMap::new();
241 for (bundle_id, child_id) in &bundle_pairs {
242 bundle_map.entry(*bundle_id).or_default().push(*child_id);
243 }
244
245 let mut export_data = Vec::new();
246 for project in &projects {
247 let items = items_by_project
248 .get(&project.id)
249 .map_or(&[][..], |v| v.as_slice());
250
251 let mut items_data = Vec::new();
252 for item in items {
253 let tag_names: Vec<&str> = tags_map
254 .get(&item.id)
255 .map(|tags| tags.iter().map(|t| t.tag_name.as_str()).collect())
256 .unwrap_or_default();
257
258 // Content-type-specific fields
259 let content_fields = match item.content() {
260 db::ContentData::Text {
261 ref body,
262 word_count,
263 reading_time_minutes,
264 } => {
265 serde_json::json!({
266 "body": body,
267 "word_count": word_count,
268 "reading_time_minutes": reading_time_minutes,
269 })
270 }
271 db::ContentData::Audio {
272 duration_seconds,
273 episode_number,
274 ..
275 } => {
276 serde_json::json!({
277 "duration_seconds": duration_seconds,
278 "episode_number": episode_number,
279 })
280 }
281 db::ContentData::Video {
282 duration_seconds,
283 width,
284 height,
285 ..
286 } => {
287 serde_json::json!({
288 "duration_seconds": duration_seconds,
289 "width": width,
290 "height": height,
291 })
292 }
293 db::ContentData::Other => serde_json::json!({}),
294 };
295
296 // Chapters (from batch map)
297 let chapters_data: Vec<serde_json::Value> = chapters_map
298 .get(&item.id)
299 .map(|chapters| {
300 chapters
301 .iter()
302 .map(|ch| {
303 serde_json::json!({
304 "title": ch.title,
305 "start_seconds": ch.start_seconds,
306 "sort_order": ch.sort_order,
307 })
308 })
309 .collect()
310 })
311 .unwrap_or_default();
312
313 // Versions (from batch map)
314 let versions_data: Vec<serde_json::Value> = versions_map
315 .get(&item.id)
316 .map(|versions| {
317 versions
318 .iter()
319 .map(|v| {
320 serde_json::json!({
321 "version_number": v.version_number,
322 "changelog": v.changelog,
323 "file_name": v.file_name,
324 "file_size_bytes": v.file_size_bytes,
325 "is_current": v.is_current,
326 "download_count": v.download_count,
327 "created_at": v.created_at,
328 })
329 })
330 .collect()
331 })
332 .unwrap_or_default();
333
334 // License keys (from batch map)
335 let license_keys_data: Vec<serde_json::Value> = license_keys_map
336 .get(&item.id)
337 .map(|keys| {
338 keys.iter()
339 .map(|lk| {
340 serde_json::json!({
341 "key_code": lk.key_code,
342 "max_activations": lk.max_activations,
343 "activation_count": lk.activation_count,
344 "revoked_at": lk.revoked_at,
345 "created_at": lk.created_at,
346 })
347 })
348 .collect()
349 })
350 .unwrap_or_default();
351
352 // Item-scoped promo codes (from batch map)
353 let item_promo_codes_data: Vec<serde_json::Value> = item_promo_codes_map
354 .get(&item.id)
355 .map(|codes| {
356 codes
357 .iter()
358 .map(|pc| {
359 serde_json::json!({
360 "code": pc.code,
361 "code_purpose": pc.code_purpose.to_string(),
362 "max_uses": pc.max_uses,
363 "use_count": pc.use_count,
364 "expires_at": pc.expires_at,
365 "created_at": pc.created_at,
366 })
367 })
368 .collect()
369 })
370 .unwrap_or_default();
371
372 let mut item_json = serde_json::json!({
373 "id": item.id,
374 "title": item.title,
375 "description": item.description,
376 "item_type": item.item_type,
377 "price_cents": item.price_cents,
378 "is_public": item.is_public,
379 "tags": tag_names,
380 "play_count": item.play_count,
381 "download_count": item.download_count,
382 "created_at": item.created_at,
383 "chapters": chapters_data,
384 "versions": versions_data,
385 "license_keys": license_keys_data,
386 "promo_codes": item_promo_codes_data,
387 });
388
389 // Merge content-type fields into item JSON
390 if let Some(obj) = content_fields.as_object() {
391 for (k, v) in obj {
392 item_json[k] = v.clone();
393 }
394 }
395
396 // Include child item IDs for bundles
397 if item.item_type == db::ItemType::Bundle
398 && let Some(child_ids) = bundle_map.get(&item.id)
399 {
400 item_json["bundle_items"] = serde_json::json!(child_ids);
401 }
402
403 items_data.push(item_json);
404 }
405
406 // Blog posts (from batch map)
407 let blog_posts_data: Vec<serde_json::Value> = blog_posts_map
408 .get(&project.id)
409 .map(|posts| {
410 posts
411 .iter()
412 .map(|post| {
413 serde_json::json!({
414 "id": post.id,
415 "title": post.title,
416 "slug": post.slug,
417 "body_markdown": post.body_markdown,
418 "published_at": post.published_at,
419 "created_at": post.created_at,
420 "updated_at": post.updated_at,
421 })
422 })
423 .collect()
424 })
425 .unwrap_or_default();
426
427 export_data.push(serde_json::json!({
428 "id": project.id,
429 "slug": project.slug,
430 "title": project.title,
431 "description": project.description,
432 "project_type": project.project_type,
433 "is_public": project.is_public,
434 "created_at": project.created_at,
435 "items": items_data,
436 "blog_posts": blog_posts_data,
437 }));
438 }
439
440 // Collections (batch-loaded to avoid N+1)
441 let collections = db::collections::get_collections_by_user(&db, user.id).await?;
442 let collection_ids: Vec<db::CollectionId> = collections.iter().map(|c| c.id).collect();
443 let collection_items_map =
444 db::collections::get_item_ids_by_collections(&db, &collection_ids).await?;
445 let mut collections_data = Vec::new();
446 for c in &collections {
447 let item_ids = collection_items_map.get(&c.id).cloned().unwrap_or_default();
448 collections_data.push(serde_json::json!({
449 "id": c.id,
450 "slug": c.slug,
451 "title": c.title,
452 "description": c.description,
453 "is_public": c.is_public,
454 "item_ids": item_ids,
455 "created_at": c.created_at,
456 }));
457 }
458
459 let custom_domain = db::custom_domains::get_custom_domain_by_user(&db, user.id).await?;
460 let custom_domain_data = custom_domain.map(|d| {
461 serde_json::json!({
462 "domain": d.domain,
463 "verified": d.verified,
464 })
465 });
466
467 let json_content = serde_json::to_string_pretty(&serde_json::json!({
468 "exported_at": chrono::Utc::now().to_rfc3339(),
469 "projects": export_data,
470 "promo_codes": promo_codes_data,
471 "collections": collections_data,
472 "custom_domain": custom_domain_data,
473 }))
474 .map_err(|e| {
475 // A full-catalog export that silently degraded to "{}" would look like a
476 // successful 200 while delivering nothing, unacceptable for a trust-
477 // critical, no-lock-in export. Surface it as a 500 instead.
478 crate::error::AppError::Internal(anyhow::anyhow!("failed to serialize project export: {e}"))
479 })?;
480
481 if is_htmx {
482 let data_uri = format!(
483 "data:application/json;charset=utf-8,{}",
484 urlencoding::encode(&json_content)
485 );
486 return Ok(ExportDownloadTemplate {
487 data_uri,
488 filename: "makenot-work-projects.json".to_string(),
489 }
490 .into_response());
491 }
492
493 download_response(
494 json_content.into_bytes(),
495 "makenot-work-projects.json",
496 "application/json",
497 )
498 }
499
500 /// Export all sales transactions as a downloadable CSV file.
501 #[tracing::instrument(skip_all, name = "exports::export_sales")]
502 pub(super) async fn export_sales(
503 State(db): State<PgPool>,
504 headers: HeaderMap,
505 AuthUser(user): AuthUser,
506 ) -> Result<Response> {
507 let is_htmx = is_htmx_request(&headers);
508 let pool = db.clone();
509 let uid = user.id;
510 let rx = spawn_paginated_csv(
511 "Date,Item ID,Item Title,Amount,Status,Buyer Email\n",
512 move |limit, offset| {
513 let pool = pool.clone();
514 async move {
515 let rows = db::transactions::get_seller_transactions_for_export_page(
516 &pool, uid, limit, offset,
517 )
518 .await?;
519 let mut buf = String::new();
520 for tx in &rows {
521 let item_title = tx.item_title.as_deref().unwrap_or("[Deleted]");
522 let item_id_str = tx
523 .item_id
524 .map_or_else(|| "[Deleted]".to_string(), |id| id.to_string());
525 let buyer_email = tx.buyer_email.as_deref().unwrap_or("");
526 writeln!(
527 buf,
528 "{},{},{},{},{},{}",
529 tx.created_at.format("%Y-%m-%d %H:%M:%S"),
530 item_id_str,
531 sanitize_csv_cell(item_title),
532 crate::formatting::format_dollars_plain(tx.amount_cents),
533 sanitize_csv_cell(&tx.status.to_string()),
534 sanitize_csv_cell(buyer_email)
535 )
536 .unwrap();
537 }
538 Ok((buf, rows.len()))
539 }
540 },
541 );
542 finish_csv(is_htmx, "makenot-work-sales.csv", rx).await
543 }
544
545 /// Export revenue splits as a downloadable CSV file.
546 #[tracing::instrument(skip_all, name = "exports::export_splits")]
547 pub(super) async fn export_splits(
548 State(db): State<PgPool>,
549 headers: HeaderMap,
550 AuthUser(user): AuthUser,
551 ) -> Result<Response> {
552 let is_htmx = is_htmx_request(&headers);
553 let pool = db.clone();
554 let uid = user.id;
555 let rx = spawn_paginated_csv(
556 "Date,Type,Direction,Recipient,Amount,Split %\n",
557 move |limit, offset| {
558 let pool = pool.clone();
559 async move {
560 let splits =
561 db::project_members::get_splits_for_export_page(&pool, uid, limit, offset)
562 .await?;
563 let mut buf = String::new();
564 for split in &splits {
565 let direction = if split.recipient_id == uid {
566 "incoming"
567 } else {
568 "outgoing"
569 };
570 writeln!(
571 buf,
572 "{},{},{},{},{},{}",
573 split.created_at.format("%Y-%m-%d %H:%M:%S"),
574 sanitize_csv_cell(&split.source_type),
575 direction,
576 sanitize_csv_cell(&split.recipient_username),
577 crate::formatting::format_dollars_plain(split.amount_cents),
578 split.split_percent,
579 )
580 .unwrap();
581 }
582 Ok((buf, splits.len()))
583 }
584 },
585 );
586 finish_csv(is_htmx, "makenot-work-splits.csv", rx).await
587 }
588
589 /// Export all purchase transactions as a downloadable CSV file.
590 #[tracing::instrument(skip_all, name = "exports::export_purchases")]
591 pub(super) async fn export_purchases(
592 State(db): State<PgPool>,
593 headers: HeaderMap,
594 AuthUser(user): AuthUser,
595 ) -> Result<Response> {
596 let is_htmx = is_htmx_request(&headers);
597 let pool = db.clone();
598 let uid = user.id;
599 let rx = spawn_paginated_csv(
600 "Date,Item ID,Item Title,Amount,Status\n",
601 move |limit, offset| {
602 let pool = pool.clone();
603 async move {
604 let transactions = db::transactions::get_buyer_transactions_for_export_page(
605 &pool, uid, limit, offset,
606 )
607 .await?;
608 // Batch-fetch titles only for this page's transactions missing
609 // the denormalized item_title.
610 let missing_title_ids: Vec<db::ItemId> = transactions
611 .iter()
612 .filter(|tx| tx.item_title.is_none())
613 .filter_map(|tx| tx.item_id)
614 .collect();
615 let title_lookup: std::collections::HashMap<db::ItemId, String> =
616 db::items::get_item_titles_batch(&pool, &missing_title_ids)
617 .await?
618 .into_iter()
619 .collect();
620
621 let mut buf = String::new();
622 for tx in &transactions {
623 let item_title = if let Some(title) = &tx.item_title {
624 title.clone()
625 } else if let Some(item_id) = tx.item_id {
626 title_lookup
627 .get(&item_id)
628 .cloned()
629 .unwrap_or_else(|| "[Deleted]".to_string())
630 } else {
631 "[Deleted]".to_string()
632 };
633 let item_id_str = tx
634 .item_id
635 .map_or_else(|| "[Deleted]".to_string(), |id| id.to_string());
636 writeln!(
637 buf,
638 "{},{},{},{},{}",
639 tx.created_at.format("%Y-%m-%d %H:%M:%S"),
640 item_id_str,
641 sanitize_csv_cell(&item_title),
642 crate::formatting::format_dollars_plain(tx.amount_cents),
643 sanitize_csv_cell(&tx.status.to_string())
644 )
645 .unwrap();
646 }
647 Ok((buf, transactions.len()))
648 }
649 },
650 );
651 finish_csv(is_htmx, "makenot-work-purchases.csv", rx).await
652 }
653
654 /// Export followers and subscribers as a downloadable CSV file.
655 #[tracing::instrument(skip_all, name = "exports::export_followers")]
656 pub(super) async fn export_followers(
657 State(db): State<PgPool>,
658 headers: HeaderMap,
659 AuthUser(user): AuthUser,
660 ) -> Result<Response> {
661 let is_htmx = is_htmx_request(&headers);
662 let pool = db.clone();
663 let uid = user.id;
664
665 // Two-section CSV (followers, then subscribers); each section pages
666 // independently so the whole thing streams in bounded batches (Run 4 S1).
667 let (tx, rx) = mpsc::channel::<Bytes>(4);
668 tokio::spawn(async move {
669 if tx
670 .send(Bytes::from_static(
671 b"Section,Username,Display Name,Email,Type,Status,Since\n",
672 ))
673 .await
674 .is_err()
675 {
676 return;
677 }
678
679 let mut offset = 0i64;
680 let mut total = 0usize;
681 loop {
682 let rows =
683 match db::follows::get_followers_for_export_page(&pool, uid, EXPORT_BATCH, offset)
684 .await
685 {
686 Ok(r) => r,
687 Err(e) => {
688 tracing::error!(error = ?e, "followers export page failed");
689 break;
690 }
691 };
692 if !rows.is_empty() {
693 let mut buf = String::new();
694 for f in &rows {
695 writeln!(
696 buf,
697 "Follower,{},{},{},{},,{}",
698 sanitize_csv_cell(&f.username),
699 sanitize_csv_cell(f.display_name.as_deref().unwrap_or("")),
700 sanitize_csv_cell(f.email.as_deref().unwrap_or("")),
701 f.target_type,
702 f.created_at.format("%Y-%m-%d %H:%M:%S"),
703 )
704 .unwrap();
705 }
706 if tx.send(Bytes::from(buf)).await.is_err() {
707 return;
708 }
709 }
710 offset += rows.len() as i64;
711 total += rows.len();
712 if (rows.len() as i64) < EXPORT_BATCH || total >= EXPORT_MAX_ROWS {
713 break;
714 }
715 }
716
717 let mut offset = 0i64;
718 let mut total = 0usize;
719 loop {
720 let rows = match db::subscriptions::get_project_subscribers_for_export_page(
721 &pool,
722 uid,
723 EXPORT_BATCH,
724 offset,
725 )
726 .await
727 {
728 Ok(r) => r,
729 Err(e) => {
730 tracing::error!(error = ?e, "subscribers export page failed");
731 break;
732 }
733 };
734 if !rows.is_empty() {
735 let mut buf = String::new();
736 for s in &rows {
737 writeln!(
738 buf,
739 "Subscriber,{},{},,{},{},{}",
740 sanitize_csv_cell(&s.username),
741 sanitize_csv_cell(s.display_name.as_deref().unwrap_or("")),
742 sanitize_csv_cell(&s.tier_name),
743 s.status,
744 s.created_at.format("%Y-%m-%d %H:%M:%S"),
745 )
746 .unwrap();
747 }
748 if tx.send(Bytes::from(buf)).await.is_err() {
749 return;
750 }
751 }
752 offset += rows.len() as i64;
753 total += rows.len();
754 if (rows.len() as i64) < EXPORT_BATCH || total >= EXPORT_MAX_ROWS {
755 break;
756 }
757 }
758 });
759 finish_csv(is_htmx, "makenot-work-followers.csv", rx).await
760 }
761
762 /// Export subscriptions as a downloadable CSV file with full detail.
763 #[tracing::instrument(skip_all, name = "exports::export_subscriptions")]
764 pub(super) async fn export_subscriptions(
765 State(db): State<PgPool>,
766 headers: HeaderMap,
767 AuthUser(user): AuthUser,
768 ) -> Result<Response> {
769 let is_htmx = is_htmx_request(&headers);
770 let pool = db.clone();
771 let uid = user.id;
772 let rx = spawn_paginated_csv(
773 "Project,Tier,Price,Username,Status,Period Start,Period End,Canceled At,Created At\n",
774 move |limit, offset| {
775 let pool = pool.clone();
776 async move {
777 let subscriptions =
778 db::subscriptions::get_subscriptions_for_export_page(&pool, uid, limit, offset)
779 .await?;
780 let fmt_opt = |dt: Option<chrono::DateTime<chrono::Utc>>| -> String {
781 dt.map(|d| d.format("%Y-%m-%d %H:%M:%S").to_string())
782 .unwrap_or_default()
783 };
784 let mut buf = String::new();
785 for s in &subscriptions {
786 writeln!(
787 buf,
788 "{},{},{},{},{},{},{},{},{}",
789 sanitize_csv_cell(&s.project_title),
790 sanitize_csv_cell(&s.tier_name),
791 crate::formatting::format_dollars_plain(s.price_cents),
792 sanitize_csv_cell(&s.username),
793 sanitize_csv_cell(&s.status.to_string()),
794 fmt_opt(s.current_period_start),
795 fmt_opt(s.current_period_end),
796 fmt_opt(s.canceled_at),
797 s.created_at.format("%Y-%m-%d %H:%M:%S"),
798 )
799 .unwrap();
800 }
801 Ok((buf, subscriptions.len()))
802 }
803 },
804 );
805 finish_csv(is_htmx, "makenot-work-subscriptions.csv", rx).await
806 }
807
808 /// Export buyer contacts (who opted to share their email) as CSV.
809 #[tracing::instrument(skip_all, name = "exports::export_contacts")]
810 pub(super) async fn export_contacts(
811 State(db): State<PgPool>,
812 headers: HeaderMap,
813 AuthUser(user): AuthUser,
814 ) -> Result<Response> {
815 let is_htmx = is_htmx_request(&headers);
816 let pool = db.clone();
817 let uid = user.id;
818 let rx = spawn_paginated_csv(
819 "Username,Email,Purchases,Total Spent,Last Purchase\n",
820 move |limit, offset| {
821 let pool = pool.clone();
822 async move {
823 let contacts =
824 db::transactions::get_seller_contacts_page(&pool, uid, limit, offset).await?;
825 let mut buf = String::new();
826 for c in &contacts {
827 writeln!(
828 buf,
829 "{},{},{},{},{}",
830 sanitize_csv_cell(&c.username),
831 sanitize_csv_cell(&c.email),
832 c.total_purchases,
833 crate::formatting::format_dollars_plain(c.total_spent_cents),
834 c.last_purchase_at.format("%Y-%m-%d"),
835 )
836 .unwrap();
837 }
838 Ok((buf, contacts.len()))
839 }
840 },
841 );
842 finish_csv(is_htmx, "makenot-work-contacts.csv", rx).await
843 }
844
845 #[cfg(test)]
846 mod tests {
847 use super::*;
848 use axum::body::to_bytes;
849 use axum::http::StatusCode;
850
851 #[test]
852 fn download_response_sets_content_type() {
853 let resp = download_response(b"hello".to_vec(), "test.csv", "text/csv").unwrap();
854 assert_eq!(resp.headers().get("Content-Type").unwrap(), "text/csv");
855 }
856
857 #[test]
858 fn download_response_sets_content_disposition() {
859 let resp = download_response(b"data".to_vec(), "export.json", "application/json").unwrap();
860 let disp = resp
861 .headers()
862 .get("Content-Disposition")
863 .unwrap()
864 .to_str()
865 .unwrap();
866 assert_eq!(disp, "attachment; filename=\"export.json\"");
867 }
868
869 #[test]
870 fn download_response_status_200() {
871 let resp = download_response(vec![], "empty.csv", "text/csv").unwrap();
872 assert_eq!(resp.status(), StatusCode::OK);
873 }
874
875 #[tokio::test]
876 async fn download_response_body_matches() {
877 let content = b"col1,col2\na,b\n".to_vec();
878 let resp = download_response(content.clone(), "f.csv", "text/csv").unwrap();
879 let body = to_bytes(resp.into_body(), 1024).await.unwrap();
880 assert_eq!(body.as_ref(), content.as_slice());
881 }
882
883 #[test]
884 fn download_response_filename_with_spaces() {
885 let resp = download_response(b"x".to_vec(), "my export.csv", "text/csv").unwrap();
886 let disp = resp
887 .headers()
888 .get("Content-Disposition")
889 .unwrap()
890 .to_str()
891 .unwrap();
892 assert!(disp.contains("my export.csv"));
893 }
894 }
895