Skip to main content

max / makenotwork

12.7 KB · 368 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::{DbBlogPost, DbItem, DbUser};
8 use crate::email::EmailClient;
9
10 /// Build the mailing-list unsubscribe URL for one subscriber: user-keyed for an
11 /// MNW account, email-keyed for an imported email-only subscriber (which has no
12 /// user id and would otherwise get no working unsubscribe link, a CAN-SPAM gap).
13 fn mailing_list_unsub_url(
14 host_url: &str,
15 subscriber: &db::mailing_lists::MailingSubscriber,
16 list_id_str: &str,
17 signing_secret: &str,
18 ) -> String {
19 use crate::email::UnsubscribeAction::MailingList;
20 match subscriber.user_id {
21 Some(uid) => crate::email::generate_unsubscribe_url(
22 host_url,
23 uid,
24 MailingList,
25 list_id_str,
26 signing_secret,
27 ),
28 None => crate::email::generate_unsubscribe_url_for_email(
29 host_url,
30 &subscriber.email,
31 MailingList,
32 list_id_str,
33 signing_secret,
34 ),
35 }
36 }
37
38 /// Spawn a bounded email fan-out off the caller's (possibly advisory-lock-held)
39 /// connection. `recipients` MUST already be bounded by the producing query's
40 /// LIMIT, this helper owns the off-lock `tokio::spawn` and the every-50
41 /// Postmark pause, so no scheduler fan-out can re-introduce an inline serial
42 /// send loop on the lock connection (Run #14 CHRONIC 2b: the shape that drifted
43 /// between the announcement and onboarding paths). `send_one` is awaited once
44 /// per recipient and owns its own per-recipient error logging.
45 ///
46 /// There is deliberately no non-spawning variant: routing every fan-out through
47 /// here is what makes "serial sends on the lock connection" unwritable.
48 fn spawn_bounded_fanout<T, F, Fut>(recipients: Vec<T>, send_one: F)
49 where
50 T: Send + 'static,
51 F: Fn(T) -> Fut + Send + 'static,
52 Fut: std::future::Future<Output = ()> + Send,
53 {
54 if recipients.is_empty() {
55 return;
56 }
57 tokio::spawn(async move {
58 for (i, recipient) in recipients.into_iter().enumerate() {
59 // Rate-limit: pause briefly every 50 emails to avoid hammering Postmark.
60 if i > 0 && i % 50 == 0 {
61 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
62 }
63 send_one(recipient).await;
64 }
65 });
66 }
67
68 /// Atomically mark an item as release-announced and send subscriber emails
69 /// via the project's content mailing list.
70 ///
71 /// Shared between the manual publish handler (`routes/api/items.rs`) and the
72 /// scheduler. Safe to call multiple times, `mark_release_announced`
73 /// is a no-op if the item was already announced.
74 #[tracing::instrument(skip_all, name = "scheduler::send_release_announcements")]
75 pub async fn send_release_announcements(
76 db: &PgPool,
77 mailer: &EmailClient,
78 config: &Config,
79 item: &DbItem,
80 ) {
81 if !db::items::mark_release_announced(db, item.id)
82 .await
83 .unwrap_or(false)
84 {
85 return;
86 }
87
88 // Skip email delivery for web-only items
89 if item.web_only {
90 return;
91 }
92
93 let Ok(Some(project)) = db::projects::get_project_by_id(db, item.project_id).await else {
94 return;
95 };
96 let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
97 return;
98 };
99 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
100 db,
101 item.project_id,
102 db::MailingListType::Content,
103 )
104 .await
105 else {
106 return;
107 };
108 let Ok(subscribers) = db::mailing_lists::get_subscribers(db, list.id).await else {
109 return;
110 };
111
112 let creator_name = creator
113 .display_name
114 .as_deref()
115 .unwrap_or(&creator.username)
116 .to_string();
117 let item_title = item.title.clone();
118 let item_url = format!("{}/i/{}", config.host_url, item.id);
119 let email_client = mailer.clone();
120 let host_url = config.host_url.clone();
121 let signing_secret = config.signing_secret.clone();
122 let list_id_str = list.id.to_string();
123
124 spawn_bounded_fanout(subscribers, move |subscriber| {
125 let email_client = email_client.clone();
126 let host_url = host_url.clone();
127 let signing_secret = signing_secret.clone();
128 let creator_name = creator_name.clone();
129 let item_title = item_title.clone();
130 let item_url = item_url.clone();
131 let list_id_str = list_id_str.clone();
132 async move {
133 let unsub_url =
134 mailing_list_unsub_url(&host_url, &subscriber, &list_id_str, &signing_secret);
135 if let Err(e) = email_client
136 .send_release_announcement(
137 &subscriber.email,
138 subscriber.display_name.as_deref(),
139 &creator_name,
140 &item_title,
141 &item_url,
142 Some(&unsub_url),
143 )
144 .await
145 {
146 tracing::error!(error = ?e, "failed to send release announcement email");
147 }
148 }
149 });
150 }
151
152 /// Atomically mark a blog post as announced and send subscriber emails
153 /// via the project's content mailing list.
154 ///
155 /// Shared between the blog post publish handlers and the scheduler.
156 /// Safe to call multiple times, `mark_blog_post_announced` is a no-op
157 /// if the post was already announced.
158 #[tracing::instrument(skip_all, name = "scheduler::send_blog_post_announcements")]
159 pub async fn send_blog_post_announcements(
160 db: &PgPool,
161 mailer: &EmailClient,
162 config: &Config,
163 post: &DbBlogPost,
164 ) {
165 if !db::blog_posts::mark_blog_post_announced(db, post.id)
166 .await
167 .unwrap_or(false)
168 {
169 return;
170 }
171
172 // Skip email delivery for web-only posts
173 if post.web_only {
174 return;
175 }
176
177 let Ok(Some(project)) = db::projects::get_project_by_id(db, post.project_id).await else {
178 return;
179 };
180 let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
181 return;
182 };
183 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
184 db,
185 post.project_id,
186 db::MailingListType::Content,
187 )
188 .await
189 else {
190 return;
191 };
192 let Ok(subscribers) = db::mailing_lists::get_subscribers(db, list.id).await else {
193 return;
194 };
195
196 let creator_name = creator
197 .display_name
198 .as_deref()
199 .unwrap_or(&creator.username)
200 .to_string();
201 let post_title = post.title.clone();
202 let post_url = format!("{}/{}/blog/{}", config.host_url, project.slug, post.slug);
203 let email_client = mailer.clone();
204 let host_url = config.host_url.clone();
205 let signing_secret = config.signing_secret.clone();
206 let list_id_str = list.id.to_string();
207
208 spawn_bounded_fanout(subscribers, move |subscriber| {
209 let email_client = email_client.clone();
210 let host_url = host_url.clone();
211 let signing_secret = signing_secret.clone();
212 let creator_name = creator_name.clone();
213 let post_title = post_title.clone();
214 let post_url = post_url.clone();
215 let list_id_str = list_id_str.clone();
216 async move {
217 let unsub_url =
218 mailing_list_unsub_url(&host_url, &subscriber, &list_id_str, &signing_secret);
219 if let Err(e) = email_client
220 .send_blog_post_announcement(
221 &subscriber.email,
222 subscriber.display_name.as_deref(),
223 &creator_name,
224 &post_title,
225 &post_url,
226 Some(&unsub_url),
227 )
228 .await
229 {
230 tracing::error!(error = ?e, "failed to send blog post announcement email");
231 }
232 }
233 });
234 }
235
236 /// Onboarding email drip steps (maps to `onboarding_email_step` i16 column).
237 #[allow(clippy::enum_variant_names)]
238 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
239 #[repr(i16)]
240 enum OnboardingStep {
241 /// Welcome email sent at signup.
242 WelcomeSent = 1,
243 /// Profile tips email (24h after welcome).
244 ProfileTipsSent = 2,
245 /// Stripe guide email (72h after welcome).
246 StripeGuideSent = 3,
247 }
248
249 impl OnboardingStep {
250 fn as_i16(self) -> i16 {
251 self as i16
252 }
253 }
254
255 /// Process the getting-started email drip sequence.
256 ///
257 /// Step 1 (welcome) is sent at signup in the auth handler.
258 /// Step 2 (profile tips) fires 24h after welcome, skipped if display_name is set.
259 /// Step 3 (Stripe guide) fires 72h after welcome, skipped if Stripe is connected.
260 ///
261 /// Only the candidate fetch + step-advance (fast DB writes) run inline; the
262 /// actual Postmark sends are spawned off the scheduler's lock-held connection
263 /// so a backlog of serial email I/O can't extend the advisory-lock hold time
264 /// (Run #14 MEDIUM, mirrors the release/blog announcement fan-out).
265 #[tracing::instrument(skip_all, name = "scheduler::send_onboarding_emails")]
266 pub(super) async fn send_onboarding_emails(db: &PgPool, mailer: &EmailClient, config: &Config) {
267 // Step 1→2: profile tips (24h after welcome)
268 let next = OnboardingStep::ProfileTipsSent;
269 if let Ok(users) = db::users::get_onboarding_candidates(
270 db,
271 OnboardingStep::WelcomeSent.as_i16(),
272 chrono::Duration::hours(24),
273 )
274 .await
275 {
276 // Batch-advance users who already set a display name (skip email)
277 let (skip, send): (Vec<_>, Vec<_>) =
278 users.into_iter().partition(|u| u.display_name.is_some());
279 advance_skipped(db, &skip, next).await;
280 claim_and_spawn_sends(db, mailer, config, send, next).await;
281 }
282
283 // Step 2→3: Stripe guide (72h after welcome)
284 let next = OnboardingStep::StripeGuideSent;
285 if let Ok(users) = db::users::get_onboarding_candidates(
286 db,
287 OnboardingStep::ProfileTipsSent.as_i16(),
288 chrono::Duration::hours(48),
289 )
290 .await
291 {
292 // Batch-advance users who already connected Stripe (skip email)
293 let (skip, send): (Vec<_>, Vec<_>) = users
294 .into_iter()
295 .partition(|u| u.stripe_account_id.is_some());
296 advance_skipped(db, &skip, next).await;
297 claim_and_spawn_sends(db, mailer, config, send, next).await;
298 }
299 }
300
301 /// Batch-advance users who don't need an email for this step (display name /
302 /// Stripe already set). Inline, a single cheap UPDATE.
303 async fn advance_skipped(db: &PgPool, skip: &[DbUser], next: OnboardingStep) {
304 if skip.is_empty() {
305 return;
306 }
307 let skip_ids: Vec<_> = skip.iter().map(|u| u.id).collect();
308 if let Err(e) = db::users::batch_advance_onboarding_step(db, &skip_ids, next.as_i16()).await {
309 tracing::warn!(count = skip_ids.len(), step = ?next, error = ?e, "failed to batch advance onboarding step");
310 }
311 }
312
313 /// Claim the send batch by advancing its step BEFORE sending (so concurrent
314 /// instances and the next tick re-exclude these users, preventing duplicate
315 /// sends), then spawn the Postmark I/O off the lock-held connection. A failed
316 /// claim leaves the rows untouched to retry next tick rather than sending
317 /// without a claim. Missing a non-critical onboarding email is better than
318 /// sending it twice.
319 async fn claim_and_spawn_sends(
320 db: &PgPool,
321 mailer: &EmailClient,
322 config: &Config,
323 send: Vec<DbUser>,
324 next: OnboardingStep,
325 ) {
326 if send.is_empty() {
327 return;
328 }
329 let send_ids: Vec<_> = send.iter().map(|u| u.id).collect();
330 if let Err(e) = db::users::batch_advance_onboarding_step(db, &send_ids, next.as_i16()).await {
331 tracing::warn!(count = send_ids.len(), step = ?next, error = ?e, "failed to claim onboarding batch; retrying next tick");
332 return;
333 }
334
335 let email_client = mailer.clone();
336 let host_url = config.host_url.clone();
337 spawn_bounded_fanout(send, move |user| {
338 let email_client = email_client.clone();
339 let host_url = host_url.clone();
340 async move {
341 let res = match next {
342 OnboardingStep::StripeGuideSent => {
343 email_client
344 .send_onboarding_stripe(
345 &user.email,
346 user.display_name.as_deref(),
347 &host_url,
348 )
349 .await
350 }
351 // ProfileTipsSent (WelcomeSent is never a "next" step).
352 _ => {
353 email_client
354 .send_onboarding_profile(
355 &user.email,
356 user.display_name.as_deref(),
357 &host_url,
358 )
359 .await
360 }
361 };
362 if let Err(e) = res {
363 tracing::error!(error = ?e, user_id = %user.id, step = ?next, "failed to send onboarding email");
364 }
365 }
366 });
367 }
368