Skip to main content

max / makenotwork

13.1 KB · 369 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 /// The unsubscribe link carried by an announcement.
14 ///
15 /// Keyed on the subscription rather than the recipient's identity, so one token
16 /// serves both jobs the surface needs: a POST unsubscribes exactly this list
17 /// (RFC 8058 one-click), and a GET opens the preferences page for everything
18 /// else they are on. It replaces the user-keyed and email-keyed forms, which
19 /// needed a different shape per recipient kind and could only ever act on the
20 /// one list.
21 fn announcement_unsub_url(
22 host_url: &str,
23 recipient: &db::lists::Recipient,
24 signing_secret: &str,
25 ) -> String {
26 crate::email::generate_subscription_unsubscribe_url(
27 host_url,
28 *recipient.subscription_id.as_uuid(),
29 signing_secret,
30 )
31 }
32
33 /// Spawn a bounded email fan-out off the caller's (possibly advisory-lock-held)
34 /// connection. `recipients` MUST already be bounded by the producing query's
35 /// LIMIT, this helper owns the off-lock `tokio::spawn` and the every-50
36 /// Postmark pause, so no scheduler fan-out can re-introduce an inline serial
37 /// send loop on the lock connection (Run #14 CHRONIC 2b: the shape that drifted
38 /// between the announcement and onboarding paths). `send_one` is awaited once
39 /// per recipient 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 /// Atomically mark an item as release-announced and send subscriber emails
64 /// via the project's content mailing list.
65 ///
66 /// Shared between the manual publish handler (`routes/api/items.rs`) and the
67 /// scheduler. Safe to call multiple times, `mark_release_announced`
68 /// is a no-op if the item was already announced.
69 #[tracing::instrument(skip_all, name = "scheduler::send_release_announcements")]
70 pub async fn send_release_announcements(
71 db: &PgPool,
72 mailer: &EmailClient,
73 config: &Config,
74 item: &DbItem,
75 ) {
76 if !db::items::mark_release_announced(db, item.id)
77 .await
78 .unwrap_or(false)
79 {
80 return;
81 }
82
83 // Skip email delivery for web-only items
84 if item.web_only {
85 return;
86 }
87
88 let Ok(Some(project)) = db::projects::get_project_by_id(db, item.project_id).await else {
89 return;
90 };
91 let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
92 return;
93 };
94 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
95 db,
96 item.project_id,
97 db::MailingListType::Content,
98 )
99 .await
100 else {
101 return;
102 };
103 let Ok(Some(unified)) = db::lists::list_for_legacy(db, list.id.into()).await else {
104 tracing::error!(list_id = %list.id, "mailing list has no unified list; skipping send");
105 return;
106 };
107 let Ok(audience) = db::lists::resolve_audience(db, unified).await else {
108 return;
109 };
110 let subscribers = audience.recipients;
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
123 spawn_bounded_fanout(subscribers, move |subscriber| {
124 let email_client = email_client.clone();
125 let host_url = host_url.clone();
126 let signing_secret = signing_secret.clone();
127 let creator_name = creator_name.clone();
128 let item_title = item_title.clone();
129 let item_url = item_url.clone();
130 async move {
131 let unsub_url = announcement_unsub_url(&host_url, &subscriber, &signing_secret);
132 if let Err(e) = email_client
133 .send_release_announcement(
134 &subscriber.email,
135 subscriber.display_name.as_deref(),
136 &creator_name,
137 &item_title,
138 &item_url,
139 Some(&unsub_url),
140 )
141 .await
142 {
143 tracing::error!(error = ?e, "failed to send release announcement email");
144 }
145 }
146 });
147 }
148
149 /// Atomically mark a blog post as announced and send subscriber emails
150 /// via the project's content mailing list.
151 ///
152 /// Shared between the blog post publish handlers and the scheduler.
153 /// Safe to call multiple times, `mark_blog_post_announced` is a no-op
154 /// if the post was already announced.
155 #[tracing::instrument(skip_all, name = "scheduler::send_blog_post_announcements")]
156 pub async fn send_blog_post_announcements(
157 db: &PgPool,
158 mailer: &EmailClient,
159 config: &Config,
160 post: &DbBlogPost,
161 ) {
162 if !db::blog_posts::mark_blog_post_announced(db, post.id)
163 .await
164 .unwrap_or(false)
165 {
166 return;
167 }
168
169 // Skip email delivery for web-only posts
170 if post.web_only {
171 return;
172 }
173
174 let Ok(Some(project)) = db::projects::get_project_by_id(db, post.project_id).await else {
175 return;
176 };
177 let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
178 return;
179 };
180 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
181 db,
182 post.project_id,
183 db::MailingListType::Content,
184 )
185 .await
186 else {
187 return;
188 };
189 let Ok(Some(unified)) = db::lists::list_for_legacy(db, list.id.into()).await else {
190 tracing::error!(list_id = %list.id, "mailing list has no unified list; skipping send");
191 return;
192 };
193 let Ok(audience) = db::lists::resolve_audience(db, unified).await else {
194 return;
195 };
196 let subscribers = audience.recipients;
197
198 let creator_name = creator
199 .display_name
200 .as_deref()
201 .unwrap_or(&creator.username)
202 .to_string();
203 let post_title = post.title.clone();
204 let post_url = format!("{}/{}/blog/{}", config.host_url, project.slug, post.slug);
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 post_title = post_title.clone();
215 let post_url = post_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_blog_post_announcement(
220 &subscriber.email,
221 subscriber.display_name.as_deref(),
222 &creator_name,
223 &post_title,
224 &post_url,
225 Some(&unsub_url),
226 )
227 .await
228 {
229 tracing::error!(error = ?e, "failed to send blog post announcement email");
230 }
231 }
232 });
233 }
234
235 /// Onboarding email drip steps (maps to `onboarding_email_step` i16 column).
236 #[allow(clippy::enum_variant_names)]
237 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
238 #[repr(i16)]
239 enum OnboardingStep {
240 /// Welcome email sent at signup.
241 WelcomeSent = 1,
242 /// Profile tips email (24h after welcome).
243 ProfileTipsSent = 2,
244 /// Stripe guide email (72h after welcome).
245 StripeGuideSent = 3,
246 }
247
248 impl OnboardingStep {
249 fn as_i16(self) -> i16 {
250 self as i16
251 }
252 }
253
254 /// Process the getting-started email drip sequence.
255 ///
256 /// Step 1 (welcome) is sent at signup in the auth handler.
257 /// Step 2 (profile tips) fires 24h after welcome, skipped if display_name is set.
258 /// Step 3 (Stripe guide) fires 72h after welcome, skipped if Stripe is connected.
259 ///
260 /// Only the candidate fetch + step-advance (fast DB writes) run inline; the
261 /// actual Postmark sends are spawned off the scheduler's lock-held connection
262 /// so a backlog of serial email I/O can't extend the advisory-lock hold time
263 /// (Run #14 MEDIUM, mirrors the release/blog announcement fan-out).
264 #[tracing::instrument(skip_all, name = "scheduler::send_onboarding_emails")]
265 pub(super) async fn send_onboarding_emails(db: &PgPool, mailer: &EmailClient, config: &Config) {
266 // Step 1→2: profile tips (24h after welcome)
267 let next = OnboardingStep::ProfileTipsSent;
268 if let Ok(users) = db::users::get_onboarding_candidates(
269 db,
270 OnboardingStep::WelcomeSent.as_i16(),
271 chrono::Duration::hours(24),
272 )
273 .await
274 {
275 // Batch-advance users who already set a display name (skip email)
276 let (skip, send): (Vec<_>, Vec<_>) =
277 users.into_iter().partition(|u| u.display_name.is_some());
278 advance_skipped(db, &skip, next).await;
279 claim_and_spawn_sends(db, mailer, config, send, next).await;
280 }
281
282 // Step 2→3: Stripe guide (72h after welcome)
283 let next = OnboardingStep::StripeGuideSent;
284 if let Ok(users) = db::users::get_onboarding_candidates(
285 db,
286 OnboardingStep::ProfileTipsSent.as_i16(),
287 chrono::Duration::hours(48),
288 )
289 .await
290 {
291 // Batch-advance users who already connected Stripe (skip email)
292 let (skip, send): (Vec<_>, Vec<_>) = users
293 .into_iter()
294 .partition(|u| u.stripe_account_id.is_some());
295 advance_skipped(db, &skip, next).await;
296 claim_and_spawn_sends(db, mailer, config, send, next).await;
297 }
298 }
299
300 /// Batch-advance users who don't need an email for this step (display name /
301 /// Stripe already set). Inline, a single cheap UPDATE.
302 async fn advance_skipped(db: &PgPool, skip: &[DbUser], next: OnboardingStep) {
303 if skip.is_empty() {
304 return;
305 }
306 let skip_ids: Vec<_> = skip.iter().map(|u| u.id).collect();
307 if let Err(e) = db::users::batch_advance_onboarding_step(db, &skip_ids, next.as_i16()).await {
308 tracing::warn!(count = skip_ids.len(), step = ?next, error = ?e, "failed to batch advance onboarding step");
309 }
310 }
311
312 /// Claim the send batch by advancing its step BEFORE sending (so concurrent
313 /// instances and the next tick re-exclude these users, preventing duplicate
314 /// sends), then spawn the Postmark I/O off the lock-held connection. A failed
315 /// claim leaves the rows untouched to retry next tick rather than sending
316 /// without a claim. Missing a non-critical onboarding email is better than
317 /// sending it twice.
318 async fn claim_and_spawn_sends(
319 db: &PgPool,
320 mailer: &EmailClient,
321 config: &Config,
322 send: Vec<DbUser>,
323 next: OnboardingStep,
324 ) {
325 if send.is_empty() {
326 return;
327 }
328 let send_ids: Vec<_> = send.iter().map(|u| u.id).collect();
329 if let Err(e) = db::users::batch_advance_onboarding_step(db, &send_ids, next.as_i16()).await {
330 tracing::warn!(count = send_ids.len(), step = ?next, error = ?e, "failed to claim onboarding batch; retrying next tick");
331 return;
332 }
333
334 let email_client = mailer.clone();
335 let host_url = config.host_url.clone();
336 spawn_bounded_fanout(send, move |user| {
337 let email_client = email_client.clone();
338 let host_url = host_url.clone();
339 async move {
340 let res = match next {
341 OnboardingStep::StripeGuideSent => {
342 email_client
343 .send_onboarding_stripe(
344 user.id,
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.id,
356 &user.email,
357 user.display_name.as_deref(),
358 &host_url,
359 )
360 .await
361 }
362 };
363 if let Err(e) = res {
364 tracing::error!(error = ?e, user_id = %user.id, step = ?next, "failed to send onboarding email");
365 }
366 }
367 });
368 }
369