Skip to main content

max / makenotwork

17.6 KB · 506 lines History Blame Raw
1 //! Release and blog post announcement emails via project mailing lists.
2
3 use sqlx::PgPool;
4
5 use crate::config::Config;
6 use crate::db;
7 use crate::db::mail_caps::Verdict;
8 use crate::db::{DbBlogPost, DbItem, DbUser, ListId};
9 use crate::email::EmailClient;
10
11 /// Build the mailing-list unsubscribe URL for one subscriber: user-keyed for an
12 /// MNW account, email-keyed for an imported email-only subscriber (which has no
13 /// user id and would otherwise get no working unsubscribe link, a CAN-SPAM gap).
14 /// The unsubscribe link carried by an announcement.
15 ///
16 /// Keyed on the subscription rather than the recipient's identity, so one token
17 /// serves both jobs the surface needs: a POST unsubscribes exactly this list
18 /// (RFC 8058 one-click), and a GET opens the preferences page for everything
19 /// else they are on. It replaces the user-keyed and email-keyed forms, which
20 /// needed a different shape per recipient kind and could only ever act on the
21 /// one list.
22 fn announcement_unsub_url(
23 host_url: &str,
24 recipient: &db::lists::Recipient,
25 signing_secret: &str,
26 ) -> String {
27 crate::email::generate_subscription_unsubscribe_url(
28 host_url,
29 *recipient.subscription_id.as_uuid(),
30 signing_secret,
31 )
32 }
33
34 /// Spawn a bounded email fan-out off the caller's (possibly advisory-lock-held)
35 /// connection. `recipients` MUST already be bounded by the producing query's
36 /// LIMIT, this helper owns the off-lock `tokio::spawn` and the every-50
37 /// Postmark pause, so no scheduler fan-out can re-introduce an inline serial
38 /// send loop on the lock connection. `send_one` is awaited once per recipient
39 /// and owns its own per-recipient error logging.
40 ///
41 /// There is deliberately no non-spawning variant: routing every fan-out through
42 /// here is what makes "serial sends on the lock connection" unwritable.
43 fn spawn_bounded_fanout<T, F, Fut>(recipients: Vec<T>, send_one: F)
44 where
45 T: Send + 'static,
46 F: Fn(T) -> Fut + Send + 'static,
47 Fut: std::future::Future<Output = ()> + Send,
48 {
49 if recipients.is_empty() {
50 return;
51 }
52 tokio::spawn(async move {
53 for (i, recipient) in recipients.into_iter().enumerate() {
54 // Rate-limit: pause briefly every 50 emails to avoid hammering Postmark.
55 if i > 0 && i % 50 == 0 {
56 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
57 }
58 send_one(recipient).await;
59 }
60 });
61 }
62
63 /// Record the fan-out about to happen, so a complaint about it can be traced
64 /// back to this creator and this list.
65 ///
66 /// `93f23f00`. `None` when the row could not be written, and the mail still
67 /// goes: an announcement is worth more than its attribution, and the failure
68 /// costs a denominator rather than a send. It shows up as a rate computed over
69 /// slightly less than was really sent, which reads high -- the safe direction
70 /// for a number that exists to warn.
71 async fn attributed(
72 db: &PgPool,
73 creator: &DbUser,
74 list_id: ListId,
75 recipients: usize,
76 ) -> Option<db::EmailSendId> {
77 match db::mail_attribution::record_send(
78 db,
79 creator.id,
80 Some(list_id),
81 db::mail_attribution::SendKind::Announcement,
82 i64::try_from(recipients).unwrap_or(i64::MAX),
83 )
84 .await
85 {
86 Ok(id) => Some(id),
87 Err(error) => {
88 tracing::warn!(error = ?error, creator_id = %creator.id,
89 "could not record the send; this fan-out will be unattributed");
90 None
91 }
92 }
93 }
94
95 /// Claim an announcement's recipients against the creator's monthly mail
96 /// allowance, or tell them why it did not go.
97 ///
98 /// The gate for both announcement fan-outs (`db::mail_caps`). It sits here
99 /// rather than inside [`spawn_bounded_fanout`] because the decision has to be
100 /// made against a *creator*, and the fan-out helper deliberately knows only
101 /// about recipients.
102 ///
103 /// Returns `false` when the send is refused, and the caller then does nothing at
104 /// all. The refusal is not silent: the creator is emailed, because this path has
105 /// no request left to answer and that mail is the whole of their notice. The
106 /// item stays marked announced either way -- re-announcing on the next scheduler
107 /// pass would mail whoever it could and drop the rest, which is the half-mailed
108 /// list the all-or-nothing reservation exists to prevent.
109 async fn allowance_admits(
110 db: &PgPool,
111 mailer: &EmailClient,
112 config: &Config,
113 creator: &DbUser,
114 what: &str,
115 recipients: usize,
116 ) -> bool {
117 let count = i64::try_from(recipients).unwrap_or(i64::MAX);
118 let verdict = match db::mail_caps::reserve(db, creator.id, count).await {
119 Ok(verdict) => verdict,
120 Err(error) => {
121 // The allowance could not be read. Sending is the safer failure:
122 // the cap protects a shared IP pool against sustained volume, and
123 // one unmetered announcement is a smaller harm than an outage in
124 // the counter silencing every creator's mail.
125 tracing::error!(error = ?error, creator_id = %creator.id,
126 "mail allowance unreadable; allowing the send");
127 return true;
128 }
129 };
130
131 let Verdict::Refused { .. } = verdict else {
132 if verdict.usage().in_warning_band() {
133 tracing::warn!(
134 creator_id = %creator.id, sent = verdict.usage().sent, cap = verdict.usage().cap,
135 "creator is inside the monthly mail warning band"
136 );
137 }
138 return true;
139 };
140
141 let explanation = verdict
142 .refusal_message()
143 .unwrap_or_else(|| "The monthly email allowance for this account is used up.".to_string());
144 tracing::warn!(
145 creator_id = %creator.id, recipients = count,
146 "announcement refused by the monthly mail allowance"
147 );
148
149 let dashboard_url = format!("{}/dashboard?tab=settings&section=creator", config.host_url);
150 if let Err(error) = mailer
151 .send_mail_cap_refusal(
152 &creator.email,
153 creator.display_name.as_deref(),
154 what,
155 &explanation,
156 &dashboard_url,
157 )
158 .await
159 {
160 tracing::error!(error = ?error, creator_id = %creator.id,
161 "failed to tell a creator their announcement was refused");
162 }
163
164 false
165 }
166
167 /// Atomically mark an item as release-announced and send subscriber emails
168 /// via the project's content mailing list.
169 ///
170 /// Shared between the item update handler, the dashboard wizard save path, and
171 /// the scheduler. Safe to call multiple times, `mark_release_announced`
172 /// is a no-op if the item was already announced.
173 #[tracing::instrument(skip_all, name = "scheduler::send_release_announcements")]
174 pub async fn send_release_announcements(
175 db: &PgPool,
176 mailer: &EmailClient,
177 config: &Config,
178 item: &DbItem,
179 ) {
180 if !db::items::mark_release_announced(db, item.id)
181 .await
182 .unwrap_or(false)
183 {
184 return;
185 }
186
187 // Skip email delivery for web-only items
188 if item.web_only {
189 return;
190 }
191
192 let Ok(Some(project)) = db::projects::get_project_by_id(db, item.project_id).await else {
193 return;
194 };
195 let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
196 return;
197 };
198 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
199 db,
200 item.project_id,
201 db::MailingListType::Content,
202 )
203 .await
204 else {
205 return;
206 };
207 let Ok(Some(unified)) = db::lists::list_for_legacy(db, list.id.into()).await else {
208 tracing::error!(list_id = %list.id, "mailing list has no unified list; skipping send");
209 return;
210 };
211 let Ok(audience) = db::lists::resolve_audience(db, unified).await else {
212 return;
213 };
214 let subscribers = audience.recipients;
215
216 if !allowance_admits(
217 db,
218 mailer,
219 config,
220 &creator,
221 &format!("Your release \"{}\"", item.title),
222 subscribers.len(),
223 )
224 .await
225 {
226 return;
227 }
228
229 let send = attributed(db, &creator, unified, subscribers.len()).await;
230 let creator_name = creator
231 .display_name
232 .as_deref()
233 .unwrap_or(&creator.username)
234 .to_string();
235 let item_title = item.title.clone();
236 let item_url = format!("{}/i/{}", config.host_url, item.id);
237 let email_client = mailer.clone();
238 let host_url = config.host_url.clone();
239 let signing_secret = config.signing_secret.clone();
240
241 spawn_bounded_fanout(subscribers, move |subscriber| {
242 let email_client = email_client.clone();
243 let host_url = host_url.clone();
244 let signing_secret = signing_secret.clone();
245 let creator_name = creator_name.clone();
246 let item_title = item_title.clone();
247 let item_url = item_url.clone();
248 async move {
249 let unsub_url = announcement_unsub_url(&host_url, &subscriber, &signing_secret);
250 if let Err(e) = email_client
251 .send_release_announcement(
252 &subscriber.email,
253 subscriber.display_name.as_deref(),
254 &creator_name,
255 &item_title,
256 &item_url,
257 crate::email::Fanout {
258 unsub_url: Some(&unsub_url),
259 send,
260 },
261 )
262 .await
263 {
264 tracing::error!(error = ?e, "failed to send release announcement email");
265 }
266 }
267 });
268 }
269
270 /// Atomically mark a blog post as announced and send subscriber emails
271 /// via the project's content mailing list.
272 ///
273 /// Shared between the blog post publish handlers and the scheduler.
274 /// Safe to call multiple times, `mark_blog_post_announced` is a no-op
275 /// if the post was already announced.
276 #[tracing::instrument(skip_all, name = "scheduler::send_blog_post_announcements")]
277 pub async fn send_blog_post_announcements(
278 db: &PgPool,
279 mailer: &EmailClient,
280 config: &Config,
281 post: &DbBlogPost,
282 ) {
283 if !db::blog_posts::mark_blog_post_announced(db, post.id)
284 .await
285 .unwrap_or(false)
286 {
287 return;
288 }
289
290 // Skip email delivery for web-only posts
291 if post.web_only {
292 return;
293 }
294
295 let Ok(Some(project)) = db::projects::get_project_by_id(db, post.project_id).await else {
296 return;
297 };
298 let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
299 return;
300 };
301 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
302 db,
303 post.project_id,
304 db::MailingListType::Content,
305 )
306 .await
307 else {
308 return;
309 };
310 let Ok(Some(unified)) = db::lists::list_for_legacy(db, list.id.into()).await else {
311 tracing::error!(list_id = %list.id, "mailing list has no unified list; skipping send");
312 return;
313 };
314 let Ok(audience) = db::lists::resolve_audience(db, unified).await else {
315 return;
316 };
317 let subscribers = audience.recipients;
318
319 if !allowance_admits(
320 db,
321 mailer,
322 config,
323 &creator,
324 &format!("Your post \"{}\"", post.title),
325 subscribers.len(),
326 )
327 .await
328 {
329 return;
330 }
331
332 let send = attributed(db, &creator, unified, subscribers.len()).await;
333 let creator_name = creator
334 .display_name
335 .as_deref()
336 .unwrap_or(&creator.username)
337 .to_string();
338 let post_title = post.title.clone();
339 let post_url = format!("{}/{}/blog/{}", config.host_url, project.slug, post.slug);
340 let email_client = mailer.clone();
341 let host_url = config.host_url.clone();
342 let signing_secret = config.signing_secret.clone();
343
344 spawn_bounded_fanout(subscribers, move |subscriber| {
345 let email_client = email_client.clone();
346 let host_url = host_url.clone();
347 let signing_secret = signing_secret.clone();
348 let creator_name = creator_name.clone();
349 let post_title = post_title.clone();
350 let post_url = post_url.clone();
351 async move {
352 let unsub_url = announcement_unsub_url(&host_url, &subscriber, &signing_secret);
353 if let Err(e) = email_client
354 .send_blog_post_announcement(
355 &subscriber.email,
356 subscriber.display_name.as_deref(),
357 &creator_name,
358 &post_title,
359 &post_url,
360 crate::email::Fanout {
361 unsub_url: Some(&unsub_url),
362 send,
363 },
364 )
365 .await
366 {
367 tracing::error!(error = ?e, "failed to send blog post announcement email");
368 }
369 }
370 });
371 }
372
373 /// Onboarding email drip steps (maps to `onboarding_email_step` i16 column).
374 #[allow(clippy::enum_variant_names)]
375 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
376 #[repr(i16)]
377 enum OnboardingStep {
378 /// Welcome email sent at signup.
379 WelcomeSent = 1,
380 /// Profile tips email (24h after welcome).
381 ProfileTipsSent = 2,
382 /// Stripe guide email (72h after welcome).
383 StripeGuideSent = 3,
384 }
385
386 impl OnboardingStep {
387 fn as_i16(self) -> i16 {
388 self as i16
389 }
390 }
391
392 /// Process the getting-started email drip sequence.
393 ///
394 /// Step 1 (welcome) is sent at signup in the auth handler.
395 /// Step 2 (profile tips) fires 24h after welcome, skipped if display_name is set.
396 /// Step 3 (Stripe guide) fires 72h after welcome, skipped if Stripe is connected.
397 ///
398 /// Only the candidate fetch + step-advance (fast DB writes) run inline; the
399 /// actual Postmark sends are spawned off the scheduler's lock-held connection
400 /// so a backlog of serial email I/O can't extend the advisory-lock hold time.
401 #[tracing::instrument(skip_all, name = "scheduler::send_onboarding_emails")]
402 pub(super) async fn send_onboarding_emails(db: &PgPool, mailer: &EmailClient, config: &Config) {
403 // Step 1→2: profile tips (24h after welcome)
404 let next = OnboardingStep::ProfileTipsSent;
405 if let Ok(users) = db::users::get_onboarding_candidates(
406 db,
407 OnboardingStep::WelcomeSent.as_i16(),
408 chrono::Duration::hours(24),
409 )
410 .await
411 {
412 // Batch-advance users who already set a display name (skip email)
413 let (skip, send): (Vec<_>, Vec<_>) =
414 users.into_iter().partition(|u| u.display_name.is_some());
415 advance_skipped(db, &skip, next).await;
416 claim_and_spawn_sends(db, mailer, config, send, next).await;
417 }
418
419 // Step 2→3: Stripe guide (72h after welcome)
420 let next = OnboardingStep::StripeGuideSent;
421 if let Ok(users) = db::users::get_onboarding_candidates(
422 db,
423 OnboardingStep::ProfileTipsSent.as_i16(),
424 chrono::Duration::hours(48),
425 )
426 .await
427 {
428 // Batch-advance users who already connected Stripe (skip email)
429 let (skip, send): (Vec<_>, Vec<_>) = users
430 .into_iter()
431 .partition(|u| u.stripe_account_id.is_some());
432 advance_skipped(db, &skip, next).await;
433 claim_and_spawn_sends(db, mailer, config, send, next).await;
434 }
435 }
436
437 /// Batch-advance users who don't need an email for this step (display name /
438 /// Stripe already set). Inline, a single cheap UPDATE.
439 async fn advance_skipped(db: &PgPool, skip: &[DbUser], next: OnboardingStep) {
440 if skip.is_empty() {
441 return;
442 }
443 let skip_ids: Vec<_> = skip.iter().map(|u| u.id).collect();
444 if let Err(e) = db::users::batch_advance_onboarding_step(db, &skip_ids, next.as_i16()).await {
445 tracing::warn!(count = skip_ids.len(), step = ?next, error = ?e, "failed to batch advance onboarding step");
446 }
447 }
448
449 /// Claim the send batch by advancing its step BEFORE sending (so concurrent
450 /// instances and the next tick re-exclude these users, preventing duplicate
451 /// sends), then spawn the Postmark I/O off the lock-held connection. A failed
452 /// claim leaves the rows untouched to retry next tick rather than sending
453 /// without a claim. Missing a non-critical onboarding email is better than
454 /// sending it twice.
455 async fn claim_and_spawn_sends(
456 db: &PgPool,
457 mailer: &EmailClient,
458 config: &Config,
459 send: Vec<DbUser>,
460 next: OnboardingStep,
461 ) {
462 if send.is_empty() {
463 return;
464 }
465 let send_ids: Vec<_> = send.iter().map(|u| u.id).collect();
466 if let Err(e) = db::users::batch_advance_onboarding_step(db, &send_ids, next.as_i16()).await {
467 tracing::warn!(count = send_ids.len(), step = ?next, error = ?e, "failed to claim onboarding batch; retrying next tick");
468 return;
469 }
470
471 let email_client = mailer.clone();
472 let host_url = config.host_url.clone();
473 spawn_bounded_fanout(send, move |user| {
474 let email_client = email_client.clone();
475 let host_url = host_url.clone();
476 async move {
477 let res = match next {
478 OnboardingStep::StripeGuideSent => {
479 email_client
480 .send_onboarding_stripe(
481 user.id,
482 &user.email,
483 user.display_name.as_deref(),
484 &host_url,
485 )
486 .await
487 }
488 // ProfileTipsSent (WelcomeSent is never a "next" step).
489 _ => {
490 email_client
491 .send_onboarding_profile(
492 user.id,
493 &user.email,
494 user.display_name.as_deref(),
495 &host_url,
496 )
497 .await
498 }
499 };
500 if let Err(e) = res {
501 tracing::error!(error = ?e, user_id = %user.id, step = ?next, "failed to send onboarding email");
502 }
503 }
504 });
505 }
506