Skip to main content

max / makenotwork

11.4 KB · 357 lines History Blame Raw
1 //! Generic import pipeline: takes an `ImportPayload` and creates MNW entities.
2 //!
3 //! Items are processed in chunks with progress updates after each chunk;
4 //! subscribers are written in batched `UNNEST` inserts (see `SUBSCRIBER_BATCH_SIZE`)
5 //! rather than one query per email. Individual item failures are logged but don't
6 //! abort the import; a hard error (e.g. mailing-list creation) propagates and the
7 //! caller (`routes/api/imports.rs`) marks the job `failed` via `fail_import_job`.
8
9 use sqlx::PgPool;
10
11 use super::{ImportPayload, ImportTier};
12 use crate::db::{self, ImportJobId, ImportJobStatus, ItemType, PriceCents, ProjectId, UserId};
13 use crate::error::Result;
14
15 /// Chunk size for progress updates.
16 const CHUNK_SIZE: usize = 50;
17
18 /// Batch size for the subscriber insert. Subscribers are written with a single
19 /// `UNNEST` INSERT per batch (not one query per email), so this is bounded by
20 /// how many parameters we want in one statement rather than progress-update
21 /// cadence, 1,000 keeps the array small while collapsing a 100k-row import
22 /// from 100k INSERTs to 100.
23 const SUBSCRIBER_BATCH_SIZE: usize = 1_000;
24
25 /// Run the full import pipeline for a job.
26 ///
27 /// Creates MNW entities (tiers, items, tags, mailing list subscribers) from the
28 /// intermediate payload. Transactions are recorded as subscriber metadata but
29 /// not inserted into the `transactions` table (no matching buyer accounts exist
30 /// for imported email addresses).
31 #[tracing::instrument(skip_all, name = "import::run_import", fields(job_id = %job_id))]
32 pub async fn run_import(
33 pool: &PgPool,
34 job_id: ImportJobId,
35 project_id: ProjectId,
36 _user_id: UserId,
37 payload: ImportPayload,
38 ) -> Result<()> {
39 db::imports::update_import_status(pool, job_id, ImportJobStatus::Processing).await?;
40
41 let mut processed: i32 = 0;
42 let mut created: i32 = 0;
43 let mut skipped: i32 = 0;
44 let mut errors: Vec<String> = Vec::new();
45
46 // --- Phase 1: Tiers ---
47 let created_tiers = import_tiers(pool, project_id, &payload.tiers, &mut errors).await;
48 processed += payload.tiers.len() as i32;
49 created += created_tiers;
50 skipped += payload.tiers.len() as i32 - created_tiers;
51 update_progress(pool, job_id, processed, created, skipped).await;
52
53 // --- Phase 2: Items ---
54 // Resolve every referenced tag slug in ONE query up front (instead of one
55 // SELECT per tag per item against the shared pool), then look tags up from
56 // the in-memory map inside the loop.
57 let all_slugs: Vec<String> = {
58 let mut set = std::collections::HashSet::new();
59 for item in &payload.items {
60 for slug in &item.tags {
61 set.insert(slug.clone());
62 }
63 }
64 set.into_iter().collect()
65 };
66 let tag_map = db::tags::get_tags_by_slugs(pool, &all_slugs)
67 .await
68 .unwrap_or_default();
69
70 for chunk in payload.items.chunks(CHUNK_SIZE) {
71 for item in chunk {
72 match import_item(pool, project_id, item, &tag_map).await {
73 Ok(true) => created += 1,
74 Ok(false) => skipped += 1,
75 Err(e) => {
76 errors.push(format!("Item '{}': {}", item.title, e));
77 skipped += 1;
78 }
79 }
80 processed += 1;
81 }
82 update_progress(pool, job_id, processed, created, skipped).await;
83 }
84
85 // --- Phase 3: Subscribers (mailing list) ---
86 // Upsert the project's content mailing list. On conflict the existing
87 // list's name is overwritten with the one passed here.
88 let list = db::mailing_lists::create_list(
89 pool,
90 project_id,
91 db::MailingListType::Content,
92 "Imported Subscribers",
93 Some("Subscribers imported from external platform"),
94 )
95 .await?;
96
97 for chunk in payload.subscribers.chunks(SUBSCRIBER_BATCH_SIZE) {
98 let emails: Vec<String> = chunk.iter().map(|s| s.email.clone()).collect();
99 match db::mailing_lists::subscribe_many_by_email(pool, list.id, &emails).await {
100 Ok(inserted) => {
101 let inserted = inserted as i32;
102 created += inserted;
103 skipped += chunk.len() as i32 - inserted;
104 }
105 Err(e) => {
106 // A batch failure is a hard DB error (not a per-row skip); surface
107 // it and count the whole chunk as skipped rather than N+1-retrying.
108 errors.push(format!("Subscriber batch ({} rows): {}", chunk.len(), e));
109 skipped += chunk.len() as i32;
110 }
111 }
112 processed += chunk.len() as i32;
113 update_progress(pool, job_id, processed, created, skipped).await;
114 }
115
116 // --- Phase 4: Transactions (count only, no DB insert) ---
117 // We can't create real transaction rows because imported buyer emails
118 // don't correspond to MNW user accounts. We just count them.
119 processed += payload.transactions.len() as i32;
120 skipped += payload.transactions.len() as i32;
121 update_progress(pool, job_id, processed, created, skipped).await;
122
123 // --- Finalize ---
124 let error_log = if errors.is_empty() {
125 None
126 } else {
127 Some(errors.join("\n"))
128 };
129
130 db::imports::complete_import_job(pool, job_id, error_log).await?;
131
132 tracing::info!(
133 processed,
134 created,
135 skipped,
136 error_count = errors.len(),
137 "import job completed"
138 );
139
140 Ok(())
141 }
142
143 /// Import subscription tiers. Returns number of tiers successfully created.
144 async fn import_tiers(
145 pool: &PgPool,
146 project_id: ProjectId,
147 tiers: &[ImportTier],
148 errors: &mut Vec<String>,
149 ) -> i32 {
150 let mut count = 0;
151 for tier in tiers {
152 let Some(price) = tier
153 .price_cents
154 .try_into()
155 .ok()
156 .and_then(|p: i32| db::PriceCents::new(p).ok())
157 else {
158 errors.push(format!(
159 "Tier '{}': price_cents {} is invalid or out of range",
160 tier.name, tier.price_cents
161 ));
162 continue;
163 };
164 match db::subscriptions::create_subscription_tier(
165 pool,
166 project_id,
167 &tier.name,
168 tier.description.as_deref(),
169 price,
170 )
171 .await
172 {
173 Ok(_) => count += 1,
174 Err(e) => {
175 // A unique violation (23505) means the tier name already exists
176 // for this project, an idempotent skip, not an error. Match on
177 // the typed SQLSTATE rather than substring-scanning the Display
178 // string (which could false-match a code appearing in a message).
179 if crate::helpers::is_unique_violation(&e) {
180 // Tier already exists, skip.
181 } else {
182 errors.push(format!("Tier '{}': {}", tier.name, e));
183 }
184 }
185 }
186 }
187 count
188 }
189
190 /// Import a single item. Returns Ok(true) if created, Ok(false) if skipped.
191 /// `tag_map` is the pre-resolved slug -> tag map (see `run_import`) so this does
192 /// no per-tag SELECT.
193 async fn import_item(
194 pool: &PgPool,
195 project_id: ProjectId,
196 item: &super::ImportItem,
197 tag_map: &std::collections::HashMap<String, db::DbTag>,
198 ) -> Result<bool> {
199 let price: i32 = item.price_cents.unwrap_or(0).try_into().map_err(|_| {
200 anyhow::anyhow!(
201 "Item '{}': price_cents {} exceeds i32 range",
202 item.title,
203 item.price_cents.unwrap_or(0)
204 )
205 })?;
206
207 // Convert HTML body to plain text (simple tag stripping for Phase A)
208 let description = item
209 .body_html
210 .as_deref()
211 .map(strip_html_tags)
212 .or(item.description.clone());
213
214 let db_item = db::items::create_item(
215 pool,
216 project_id,
217 &item.title,
218 description.as_deref(),
219 PriceCents::from_db(price),
220 ItemType::Digital, // Default type for imports
221 db::AiTier::Handmade,
222 None,
223 )
224 .await?;
225
226 // Publish if the source item was public
227 if item.is_public {
228 sqlx::query(
229 "UPDATE items SET published = true, published_at = COALESCE($2, NOW()) WHERE id = $1",
230 )
231 .bind(db_item.id)
232 .bind(item.published_at)
233 .execute(pool)
234 .await?;
235 }
236
237 // Attach tags via the pre-resolved map (dot-notation, e.g.
238 // "audio.genre.electronic"), no per-tag SELECT.
239 for tag_slug in &item.tags {
240 if let Some(tag) = tag_map.get(tag_slug) {
241 let _ = db::tags::add_tag_to_item(pool, db_item.id, tag.id, false).await;
242 }
243 }
244
245 Ok(true)
246 }
247
248 /// Minimal HTML tag stripper. Converts <br>, <p>, <li> to newlines,
249 /// strips all other tags, and collapses whitespace.
250 fn strip_html_tags(html: &str) -> String {
251 let mut result = String::with_capacity(html.len());
252 let mut in_tag = false;
253 let mut last_was_newline = false;
254
255 // Replace block-level tags with newlines first
256 let html = html
257 .replace("<br>", "\n")
258 .replace("<br/>", "\n")
259 .replace("<br />", "\n")
260 .replace("</p>", "\n")
261 .replace("</li>", "\n")
262 .replace("</div>", "\n");
263
264 for ch in html.chars() {
265 if ch == '<' {
266 in_tag = true;
267 continue;
268 }
269 if ch == '>' {
270 in_tag = false;
271 continue;
272 }
273 if !in_tag {
274 if ch == '\n' {
275 if !last_was_newline {
276 result.push('\n');
277 last_was_newline = true;
278 }
279 } else {
280 result.push(ch);
281 last_was_newline = false;
282 }
283 }
284 }
285
286 // Cap output length to guard against multi-MB HTML bodies.
287 const MAX_STRIPPED_LEN: usize = 512 * 1024; // 512 KB
288 let trimmed = result.trim();
289 if trimmed.len() > MAX_STRIPPED_LEN {
290 // Truncate on a char boundary
291 let mut end = MAX_STRIPPED_LEN;
292 while !trimmed.is_char_boundary(end) {
293 end -= 1;
294 }
295 trimmed[..end].to_string()
296 } else {
297 trimmed.to_string()
298 }
299 }
300
301 /// Update progress on the import job, and refresh its liveness heartbeat so the
302 /// reaper doesn't mistake a long-but-progressing import for a crashed one.
303 async fn update_progress(
304 pool: &PgPool,
305 job_id: ImportJobId,
306 processed: i32,
307 created: i32,
308 skipped: i32,
309 ) {
310 if let Err(e) =
311 db::imports::update_import_progress(pool, job_id, processed, created, skipped).await
312 {
313 tracing::warn!(error = %e, "failed to update import progress");
314 }
315 if let Err(e) = db::imports::bump_import_heartbeat(pool, job_id).await {
316 tracing::warn!(error = %e, "failed to bump import heartbeat");
317 }
318 }
319
320 #[cfg(test)]
321 mod tests {
322 use super::*;
323
324 #[test]
325 fn strip_html_basic() {
326 assert_eq!(strip_html_tags("<p>Hello</p>"), "Hello");
327 }
328
329 #[test]
330 fn strip_html_br_tags() {
331 assert_eq!(strip_html_tags("line1<br>line2"), "line1\nline2");
332 assert_eq!(strip_html_tags("line1<br/>line2"), "line1\nline2");
333 }
334
335 #[test]
336 fn strip_html_nested() {
337 assert_eq!(
338 strip_html_tags("<div><p>Hello <strong>world</strong></p></div>"),
339 "Hello world"
340 );
341 }
342
343 #[test]
344 fn strip_html_collapses_newlines() {
345 assert_eq!(
346 strip_html_tags("<p>One</p><p>Two</p><p>Three</p>"),
347 "One\nTwo\nThree"
348 );
349 }
350
351 #[test]
352 fn strip_html_empty() {
353 assert_eq!(strip_html_tags(""), "");
354 assert_eq!(strip_html_tags("<br><br><br>"), "");
355 }
356 }
357