Skip to main content

max / makenotwork

35.8 KB · 956 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 one item's sales as a downloadable CSV file.
546 ///
547 /// `1ea96868`. This existed as `static/tab-item-sales.js`, which built the file
548 /// in the browser by scraping the rendered table and quoting each cell with a
549 /// bare `"`. That neutralised no formula prefix and escaped no embedded quote,
550 /// and the Buyer column is `guest_email` -- typed by the buyer at guest
551 /// checkout, so attacker-controlled and landing in a file the creator opens.
552 /// Every server-side export in this module already ran `sanitize_csv_cell`;
553 /// that one was the only export in the tree built client-side and inherited
554 /// none of it.
555 ///
556 /// # Why it reads the whole set rather than paging
557 ///
558 /// The item's own sales tab already calls `get_sales_by_item` and renders every
559 /// row, so holding the same set here adds no exposure the page did not have.
560 /// Paging it would want a second query, and this one is item-scoped rather than
561 /// seller-scoped: it is a page of a creator's history, not the history.
562 ///
563 /// Ownership is the query's, not a separate check: `get_sales_by_item` takes
564 /// `seller_id` and filters on it, so another creator's item id returns nothing.
565 #[tracing::instrument(skip_all, name = "exports::export_item_sales")]
566 pub(super) async fn export_item_sales(
567 State(db): State<PgPool>,
568 headers: HeaderMap,
569 AuthUser(user): AuthUser,
570 axum::extract::Path(item_id): axum::extract::Path<crate::db::ItemId>,
571 ) -> Result<Response> {
572 let is_htmx = is_htmx_request(&headers);
573 let sales = db::transactions::get_sales_by_item(&db, item_id, user.id).await?;
574
575 let mut body = String::new();
576 for tx in &sales {
577 let buyer = tx
578 .guest_email
579 .clone()
580 .or_else(|| tx.buyer_id.map(|_| "Registered user".to_string()))
581 .unwrap_or_else(|| "Unknown".to_string());
582 writeln!(
583 body,
584 "{},{},{},{}",
585 tx.created_at.format("%Y-%m-%d %H:%M"),
586 sanitize_csv_cell(&buyer),
587 crate::formatting::format_dollars_plain(tx.amount_cents.as_i64()),
588 sanitize_csv_cell(&tx.status.to_string()),
589 )
590 .unwrap();
591 }
592
593 // One page, then done. `spawn_paginated_csv` ends when a page comes back
594 // shorter than a batch, so the row count is what terminates it and an
595 // oversized set still stops on the empty second page.
596 let rows = sales.len();
597 let mut once = Some(body);
598 let rx = spawn_paginated_csv("Date,Buyer,Amount,Status\n", move |_limit, _offset| {
599 let page = once.take();
600 async move { Ok(page.map_or_else(|| (String::new(), 0), |text| (text, rows))) }
601 });
602
603 finish_csv(is_htmx, "makenot-work-item-sales.csv", rx).await
604 }
605
606 /// Export revenue splits as a downloadable CSV file.
607 #[tracing::instrument(skip_all, name = "exports::export_splits")]
608 pub(super) async fn export_splits(
609 State(db): State<PgPool>,
610 headers: HeaderMap,
611 AuthUser(user): AuthUser,
612 ) -> Result<Response> {
613 let is_htmx = is_htmx_request(&headers);
614 let pool = db.clone();
615 let uid = user.id;
616 let rx = spawn_paginated_csv(
617 "Date,Type,Direction,Recipient,Amount,Split %\n",
618 move |limit, offset| {
619 let pool = pool.clone();
620 async move {
621 let splits =
622 db::project_members::get_splits_for_export_page(&pool, uid, limit, offset)
623 .await?;
624 let mut buf = String::new();
625 for split in &splits {
626 let direction = if split.recipient_id == uid {
627 "incoming"
628 } else {
629 "outgoing"
630 };
631 writeln!(
632 buf,
633 "{},{},{},{},{},{}",
634 split.created_at.format("%Y-%m-%d %H:%M:%S"),
635 sanitize_csv_cell(&split.source_type),
636 direction,
637 sanitize_csv_cell(&split.recipient_username),
638 crate::formatting::format_dollars_plain(split.amount_cents),
639 split.split_percent,
640 )
641 .unwrap();
642 }
643 Ok((buf, splits.len()))
644 }
645 },
646 );
647 finish_csv(is_htmx, "makenot-work-splits.csv", rx).await
648 }
649
650 /// Export all purchase transactions as a downloadable CSV file.
651 #[tracing::instrument(skip_all, name = "exports::export_purchases")]
652 pub(super) async fn export_purchases(
653 State(db): State<PgPool>,
654 headers: HeaderMap,
655 AuthUser(user): AuthUser,
656 ) -> Result<Response> {
657 let is_htmx = is_htmx_request(&headers);
658 let pool = db.clone();
659 let uid = user.id;
660 let rx = spawn_paginated_csv(
661 "Date,Item ID,Item Title,Amount,Status\n",
662 move |limit, offset| {
663 let pool = pool.clone();
664 async move {
665 let transactions = db::transactions::get_buyer_transactions_for_export_page(
666 &pool, uid, limit, offset,
667 )
668 .await?;
669 // Batch-fetch titles only for this page's transactions missing
670 // the denormalized item_title.
671 let missing_title_ids: Vec<db::ItemId> = transactions
672 .iter()
673 .filter(|tx| tx.item_title.is_none())
674 .filter_map(|tx| tx.item_id)
675 .collect();
676 let title_lookup: std::collections::HashMap<db::ItemId, String> =
677 db::items::get_item_titles_batch(&pool, &missing_title_ids)
678 .await?
679 .into_iter()
680 .collect();
681
682 let mut buf = String::new();
683 for tx in &transactions {
684 let item_title = if let Some(title) = &tx.item_title {
685 title.clone()
686 } else if let Some(item_id) = tx.item_id {
687 title_lookup
688 .get(&item_id)
689 .cloned()
690 .unwrap_or_else(|| "[Deleted]".to_string())
691 } else {
692 "[Deleted]".to_string()
693 };
694 let item_id_str = tx
695 .item_id
696 .map_or_else(|| "[Deleted]".to_string(), |id| id.to_string());
697 writeln!(
698 buf,
699 "{},{},{},{},{}",
700 tx.created_at.format("%Y-%m-%d %H:%M:%S"),
701 item_id_str,
702 sanitize_csv_cell(&item_title),
703 crate::formatting::format_dollars_plain(tx.amount_cents),
704 sanitize_csv_cell(&tx.status.to_string())
705 )
706 .unwrap();
707 }
708 Ok((buf, transactions.len()))
709 }
710 },
711 );
712 finish_csv(is_htmx, "makenot-work-purchases.csv", rx).await
713 }
714
715 /// Export followers and subscribers as a downloadable CSV file.
716 #[tracing::instrument(skip_all, name = "exports::export_followers")]
717 pub(super) async fn export_followers(
718 State(db): State<PgPool>,
719 headers: HeaderMap,
720 AuthUser(user): AuthUser,
721 ) -> Result<Response> {
722 let is_htmx = is_htmx_request(&headers);
723 let pool = db.clone();
724 let uid = user.id;
725
726 // Two-section CSV (followers, then subscribers); each section pages
727 // independently so the whole thing streams in bounded batches (Run 4 S1).
728 let (tx, rx) = mpsc::channel::<Bytes>(4);
729 tokio::spawn(async move {
730 if tx
731 .send(Bytes::from_static(
732 b"Section,Username,Display Name,Email,Type,Status,Since\n",
733 ))
734 .await
735 .is_err()
736 {
737 return;
738 }
739
740 let mut offset = 0i64;
741 let mut total = 0usize;
742 loop {
743 let rows =
744 match db::follows::get_followers_for_export_page(&pool, uid, EXPORT_BATCH, offset)
745 .await
746 {
747 Ok(r) => r,
748 Err(e) => {
749 tracing::error!(error = ?e, "followers export page failed");
750 break;
751 }
752 };
753 if !rows.is_empty() {
754 let mut buf = String::new();
755 for f in &rows {
756 writeln!(
757 buf,
758 "Follower,{},{},{},{},,{}",
759 sanitize_csv_cell(&f.username),
760 sanitize_csv_cell(f.display_name.as_deref().unwrap_or("")),
761 sanitize_csv_cell(f.email.as_deref().unwrap_or("")),
762 f.target_type,
763 f.created_at.format("%Y-%m-%d %H:%M:%S"),
764 )
765 .unwrap();
766 }
767 if tx.send(Bytes::from(buf)).await.is_err() {
768 return;
769 }
770 }
771 offset += rows.len() as i64;
772 total += rows.len();
773 if (rows.len() as i64) < EXPORT_BATCH || total >= EXPORT_MAX_ROWS {
774 break;
775 }
776 }
777
778 let mut offset = 0i64;
779 let mut total = 0usize;
780 loop {
781 let rows = match db::subscriptions::get_project_subscribers_for_export_page(
782 &pool,
783 uid,
784 EXPORT_BATCH,
785 offset,
786 )
787 .await
788 {
789 Ok(r) => r,
790 Err(e) => {
791 tracing::error!(error = ?e, "subscribers export page failed");
792 break;
793 }
794 };
795 if !rows.is_empty() {
796 let mut buf = String::new();
797 for s in &rows {
798 writeln!(
799 buf,
800 "Subscriber,{},{},,{},{},{}",
801 sanitize_csv_cell(&s.username),
802 sanitize_csv_cell(s.display_name.as_deref().unwrap_or("")),
803 sanitize_csv_cell(&s.tier_name),
804 s.status,
805 s.created_at.format("%Y-%m-%d %H:%M:%S"),
806 )
807 .unwrap();
808 }
809 if tx.send(Bytes::from(buf)).await.is_err() {
810 return;
811 }
812 }
813 offset += rows.len() as i64;
814 total += rows.len();
815 if (rows.len() as i64) < EXPORT_BATCH || total >= EXPORT_MAX_ROWS {
816 break;
817 }
818 }
819 });
820 finish_csv(is_htmx, "makenot-work-followers.csv", rx).await
821 }
822
823 /// Export subscriptions as a downloadable CSV file with full detail.
824 #[tracing::instrument(skip_all, name = "exports::export_subscriptions")]
825 pub(super) async fn export_subscriptions(
826 State(db): State<PgPool>,
827 headers: HeaderMap,
828 AuthUser(user): AuthUser,
829 ) -> Result<Response> {
830 let is_htmx = is_htmx_request(&headers);
831 let pool = db.clone();
832 let uid = user.id;
833 let rx = spawn_paginated_csv(
834 "Project,Tier,Price,Username,Status,Period Start,Period End,Canceled At,Created At\n",
835 move |limit, offset| {
836 let pool = pool.clone();
837 async move {
838 let subscriptions =
839 db::subscriptions::get_subscriptions_for_export_page(&pool, uid, limit, offset)
840 .await?;
841 let fmt_opt = |dt: Option<chrono::DateTime<chrono::Utc>>| -> String {
842 dt.map(|d| d.format("%Y-%m-%d %H:%M:%S").to_string())
843 .unwrap_or_default()
844 };
845 let mut buf = String::new();
846 for s in &subscriptions {
847 writeln!(
848 buf,
849 "{},{},{},{},{},{},{},{},{}",
850 sanitize_csv_cell(&s.project_title),
851 sanitize_csv_cell(&s.tier_name),
852 crate::formatting::format_dollars_plain(s.price_cents),
853 sanitize_csv_cell(&s.username),
854 sanitize_csv_cell(&s.status.to_string()),
855 fmt_opt(s.current_period_start),
856 fmt_opt(s.current_period_end),
857 fmt_opt(s.canceled_at),
858 s.created_at.format("%Y-%m-%d %H:%M:%S"),
859 )
860 .unwrap();
861 }
862 Ok((buf, subscriptions.len()))
863 }
864 },
865 );
866 finish_csv(is_htmx, "makenot-work-subscriptions.csv", rx).await
867 }
868
869 /// Export buyer contacts (who opted to share their email) as CSV.
870 #[tracing::instrument(skip_all, name = "exports::export_contacts")]
871 pub(super) async fn export_contacts(
872 State(db): State<PgPool>,
873 headers: HeaderMap,
874 AuthUser(user): AuthUser,
875 ) -> Result<Response> {
876 let is_htmx = is_htmx_request(&headers);
877 let pool = db.clone();
878 let uid = user.id;
879 let rx = spawn_paginated_csv(
880 "Username,Email,Purchases,Total Spent,Last Purchase\n",
881 move |limit, offset| {
882 let pool = pool.clone();
883 async move {
884 let contacts =
885 db::transactions::get_seller_contacts_page(&pool, uid, limit, offset).await?;
886 let mut buf = String::new();
887 for c in &contacts {
888 writeln!(
889 buf,
890 "{},{},{},{},{}",
891 sanitize_csv_cell(&c.username),
892 sanitize_csv_cell(&c.email),
893 c.total_purchases,
894 crate::formatting::format_dollars_plain(c.total_spent_cents),
895 c.last_purchase_at.format("%Y-%m-%d"),
896 )
897 .unwrap();
898 }
899 Ok((buf, contacts.len()))
900 }
901 },
902 );
903 finish_csv(is_htmx, "makenot-work-contacts.csv", rx).await
904 }
905
906 #[cfg(test)]
907 mod tests {
908 use super::*;
909 use axum::body::to_bytes;
910 use axum::http::StatusCode;
911
912 #[test]
913 fn download_response_sets_content_type() {
914 let resp = download_response(b"hello".to_vec(), "test.csv", "text/csv").unwrap();
915 assert_eq!(resp.headers().get("Content-Type").unwrap(), "text/csv");
916 }
917
918 #[test]
919 fn download_response_sets_content_disposition() {
920 let resp = download_response(b"data".to_vec(), "export.json", "application/json").unwrap();
921 let disp = resp
922 .headers()
923 .get("Content-Disposition")
924 .unwrap()
925 .to_str()
926 .unwrap();
927 assert_eq!(disp, "attachment; filename=\"export.json\"");
928 }
929
930 #[test]
931 fn download_response_status_200() {
932 let resp = download_response(vec![], "empty.csv", "text/csv").unwrap();
933 assert_eq!(resp.status(), StatusCode::OK);
934 }
935
936 #[tokio::test]
937 async fn download_response_body_matches() {
938 let content = b"col1,col2\na,b\n".to_vec();
939 let resp = download_response(content.clone(), "f.csv", "text/csv").unwrap();
940 let body = to_bytes(resp.into_body(), 1024).await.unwrap();
941 assert_eq!(body.as_ref(), content.as_slice());
942 }
943
944 #[test]
945 fn download_response_filename_with_spaces() {
946 let resp = download_response(b"x".to_vec(), "my export.csv", "text/csv").unwrap();
947 let disp = resp
948 .headers()
949 .get("Content-Disposition")
950 .unwrap()
951 .to_str()
952 .unwrap();
953 assert!(disp.contains("my export.csv"));
954 }
955 }
956