Skip to main content

max / makenotwork

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