Skip to main content

max / makenotwork

11.4 KB · 356 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 // Ensure the project has a content mailing list
87 let list = db::mailing_lists::create_list(
88 pool,
89 project_id,
90 db::MailingListType::Content,
91 "Imported Subscribers",
92 Some("Subscribers imported from external platform"),
93 )
94 .await?;
95
96 for chunk in payload.subscribers.chunks(SUBSCRIBER_BATCH_SIZE) {
97 let emails: Vec<String> = chunk.iter().map(|s| s.email.clone()).collect();
98 match db::mailing_lists::subscribe_many_by_email(pool, list.id, &emails).await {
99 Ok(inserted) => {
100 let inserted = inserted as i32;
101 created += inserted;
102 skipped += chunk.len() as i32 - inserted;
103 }
104 Err(e) => {
105 // A batch failure is a hard DB error (not a per-row skip); surface
106 // it and count the whole chunk as skipped rather than N+1-retrying.
107 errors.push(format!("Subscriber batch ({} rows): {}", chunk.len(), e));
108 skipped += chunk.len() as i32;
109 }
110 }
111 processed += chunk.len() as i32;
112 update_progress(pool, job_id, processed, created, skipped).await;
113 }
114
115 // ── Phase 4: Transactions (count only, no DB insert) ──
116 // We can't create real transaction rows because imported buyer emails
117 // don't correspond to MNW user accounts. We just count them.
118 processed += payload.transactions.len() as i32;
119 skipped += payload.transactions.len() as i32;
120 update_progress(pool, job_id, processed, created, skipped).await;
121
122 // ── Finalize ──
123 let error_log = if errors.is_empty() {
124 None
125 } else {
126 Some(errors.join("\n"))
127 };
128
129 db::imports::complete_import_job(pool, job_id, error_log).await?;
130
131 tracing::info!(
132 processed,
133 created,
134 skipped,
135 error_count = errors.len(),
136 "import job completed"
137 );
138
139 Ok(())
140 }
141
142 /// Import subscription tiers. Returns number of tiers successfully created.
143 async fn import_tiers(
144 pool: &PgPool,
145 project_id: ProjectId,
146 tiers: &[ImportTier],
147 errors: &mut Vec<String>,
148 ) -> i32 {
149 let mut count = 0;
150 for tier in tiers {
151 let Some(price) = tier
152 .price_cents
153 .try_into()
154 .ok()
155 .and_then(|p: i32| db::PriceCents::new(p).ok())
156 else {
157 errors.push(format!(
158 "Tier '{}': price_cents {} is invalid or out of range",
159 tier.name, tier.price_cents
160 ));
161 continue;
162 };
163 match db::subscriptions::create_subscription_tier(
164 pool,
165 project_id,
166 &tier.name,
167 tier.description.as_deref(),
168 price,
169 )
170 .await
171 {
172 Ok(_) => count += 1,
173 Err(e) => {
174 // A unique violation (23505) means the tier name already exists
175 // for this project, an idempotent skip, not an error. Match on
176 // the typed SQLSTATE rather than substring-scanning the Display
177 // string (which could false-match a code appearing in a message).
178 if crate::helpers::is_unique_violation(&e) {
179 // Tier already exists, skip.
180 } else {
181 errors.push(format!("Tier '{}': {}", tier.name, e));
182 }
183 }
184 }
185 }
186 count
187 }
188
189 /// Import a single item. Returns Ok(true) if created, Ok(false) if skipped.
190 /// `tag_map` is the pre-resolved slug -> tag map (see `run_import`) so this does
191 /// no per-tag SELECT.
192 async fn import_item(
193 pool: &PgPool,
194 project_id: ProjectId,
195 item: &super::ImportItem,
196 tag_map: &std::collections::HashMap<String, db::DbTag>,
197 ) -> Result<bool> {
198 let price: i32 = item.price_cents.unwrap_or(0).try_into().map_err(|_| {
199 anyhow::anyhow!(
200 "Item '{}': price_cents {} exceeds i32 range",
201 item.title,
202 item.price_cents.unwrap_or(0)
203 )
204 })?;
205
206 // Convert HTML body to plain text (simple tag stripping for Phase A)
207 let description = item
208 .body_html
209 .as_deref()
210 .map(strip_html_tags)
211 .or(item.description.clone());
212
213 let db_item = db::items::create_item(
214 pool,
215 project_id,
216 &item.title,
217 description.as_deref(),
218 PriceCents::from_db(price),
219 ItemType::Digital, // Default type for imports
220 db::AiTier::Handmade,
221 None,
222 )
223 .await?;
224
225 // Publish if the source item was public
226 if item.is_public {
227 sqlx::query(
228 "UPDATE items SET published = true, published_at = COALESCE($2, NOW()) WHERE id = $1",
229 )
230 .bind(db_item.id)
231 .bind(item.published_at)
232 .execute(pool)
233 .await?;
234 }
235
236 // Attach tags via the pre-resolved map (dot-notation, e.g.
237 // "audio.genre.electronic"), no per-tag SELECT.
238 for tag_slug in &item.tags {
239 if let Some(tag) = tag_map.get(tag_slug) {
240 let _ = db::tags::add_tag_to_item(pool, db_item.id, tag.id, false).await;
241 }
242 }
243
244 Ok(true)
245 }
246
247 /// Minimal HTML tag stripper. Converts <br>, <p>, <li> to newlines,
248 /// strips all other tags, and collapses whitespace.
249 fn strip_html_tags(html: &str) -> String {
250 let mut result = String::with_capacity(html.len());
251 let mut in_tag = false;
252 let mut last_was_newline = false;
253
254 // Replace block-level tags with newlines first
255 let html = html
256 .replace("<br>", "\n")
257 .replace("<br/>", "\n")
258 .replace("<br />", "\n")
259 .replace("</p>", "\n")
260 .replace("</li>", "\n")
261 .replace("</div>", "\n");
262
263 for ch in html.chars() {
264 if ch == '<' {
265 in_tag = true;
266 continue;
267 }
268 if ch == '>' {
269 in_tag = false;
270 continue;
271 }
272 if !in_tag {
273 if ch == '\n' {
274 if !last_was_newline {
275 result.push('\n');
276 last_was_newline = true;
277 }
278 } else {
279 result.push(ch);
280 last_was_newline = false;
281 }
282 }
283 }
284
285 // Cap output length to guard against multi-MB HTML bodies.
286 const MAX_STRIPPED_LEN: usize = 512 * 1024; // 512 KB
287 let trimmed = result.trim();
288 if trimmed.len() > MAX_STRIPPED_LEN {
289 // Truncate on a char boundary
290 let mut end = MAX_STRIPPED_LEN;
291 while !trimmed.is_char_boundary(end) {
292 end -= 1;
293 }
294 trimmed[..end].to_string()
295 } else {
296 trimmed.to_string()
297 }
298 }
299
300 /// Update progress on the import job, and refresh its liveness heartbeat so the
301 /// reaper doesn't mistake a long-but-progressing import for a crashed one.
302 async fn update_progress(
303 pool: &PgPool,
304 job_id: ImportJobId,
305 processed: i32,
306 created: i32,
307 skipped: i32,
308 ) {
309 if let Err(e) =
310 db::imports::update_import_progress(pool, job_id, processed, created, skipped).await
311 {
312 tracing::warn!(error = %e, "failed to update import progress");
313 }
314 if let Err(e) = db::imports::bump_import_heartbeat(pool, job_id).await {
315 tracing::warn!(error = %e, "failed to bump import heartbeat");
316 }
317 }
318
319 #[cfg(test)]
320 mod tests {
321 use super::*;
322
323 #[test]
324 fn strip_html_basic() {
325 assert_eq!(strip_html_tags("<p>Hello</p>"), "Hello");
326 }
327
328 #[test]
329 fn strip_html_br_tags() {
330 assert_eq!(strip_html_tags("line1<br>line2"), "line1\nline2");
331 assert_eq!(strip_html_tags("line1<br/>line2"), "line1\nline2");
332 }
333
334 #[test]
335 fn strip_html_nested() {
336 assert_eq!(
337 strip_html_tags("<div><p>Hello <strong>world</strong></p></div>"),
338 "Hello world"
339 );
340 }
341
342 #[test]
343 fn strip_html_collapses_newlines() {
344 assert_eq!(
345 strip_html_tags("<p>One</p><p>Two</p><p>Three</p>"),
346 "One\nTwo\nThree"
347 );
348 }
349
350 #[test]
351 fn strip_html_empty() {
352 assert_eq!(strip_html_tags(""), "");
353 assert_eq!(strip_html_tags("<br><br><br>"), "");
354 }
355 }
356