Skip to main content

max / makenotwork

33.6 KB · 954 lines History Blame Raw
1 //! One subscription model for every list-like thing.
2 //!
3 //! Step 2 of the plan in the maintainer wiki (`mnw-mailing-lists`). Three
4 //! tables: a `list` is anything somebody can be subscribed to, a
5 //! `list_subscription` is one recipient's relationship to one list, and a
6 //! `consent_event` is why we believe we are allowed to mail them.
7 //!
8 //! **Nothing sends through this yet.** `mailing_lists` and `email_signups`
9 //! remain authoritative until the resolver lands (step 3). This module exists
10 //! so the schema has typed access and the backfill has tests, not so callers
11 //! can start using it piecemeal, which is how the five parallel mechanisms this
12 //! replaces came to exist in the first place.
13 //! <!-- wiki: mnw-mailing-lists -->
14
15 use sqlx::PgPool;
16
17 use super::enums::{ConsentEvent, ListKind, ListScope, SubscriptionSource, SubscriptionState};
18 use super::id_types::{ListId, ListSubscriptionId, UserId};
19 use crate::error::Result;
20
21 /// A subscriber: an account or a bare address, never both. The old
22 /// `mailing_list_subscribers` allowed both at once and left "which one is
23 /// authoritative" to whoever read the row next; the new table's CHECK makes
24 /// that unrepresentable, and this type carries the same rule into Rust.
25 #[derive(Debug, Clone)]
26 pub enum Subscriber {
27 User(UserId),
28 Email(String),
29 }
30
31 /// Find a list by what it is attached to.
32 ///
33 /// `scope_id` must be `None` for [`ListScope::Platform`] and `Some` otherwise;
34 /// the database CHECK rejects the other combinations, so a mismatch here
35 /// returns no row rather than the wrong one.
36 #[tracing::instrument(skip_all)]
37 pub async fn find_list(
38 pool: &PgPool,
39 scope: ListScope,
40 scope_id: Option<uuid::Uuid>,
41 kind: ListKind,
42 ) -> Result<Option<ListId>> {
43 let id = sqlx::query_scalar::<_, ListId>(
44 "SELECT id FROM lists \
45 WHERE scope = $1 AND kind = $2 \
46 AND (scope_id = $3 OR ($3::uuid IS NULL AND scope_id IS NULL))",
47 )
48 .bind(scope.to_string())
49 .bind(kind.to_string())
50 .bind(scope_id)
51 .fetch_optional(pool)
52 .await?;
53 Ok(id)
54 }
55
56 /// Record a subscription and the consent event that justifies it, in one
57 /// transaction.
58 ///
59 /// The two are written together because a subscription without its consent
60 /// event is exactly the state this whole model exists to eliminate: someone on
61 /// a list with no record of why. Re-subscribing an address that had left moves
62 /// it back and appends a fresh event rather than editing the old one.
63 #[tracing::instrument(skip_all)]
64 pub async fn subscribe(
65 pool: &PgPool,
66 list_id: ListId,
67 subscriber: &Subscriber,
68 state: SubscriptionState,
69 source: SubscriptionSource,
70 event: ConsentEvent,
71 evidence: Option<&str>,
72 ) -> Result<ListSubscriptionId> {
73 let (user_id, email) = match subscriber {
74 Subscriber::User(id) => (Some(*id), None),
75 Subscriber::Email(addr) => (None, Some(addr.to_lowercase())),
76 };
77 let confirmed_at = (state == SubscriptionState::Confirmed).then(chrono::Utc::now);
78 // Kept in step with `state` here rather than at each call site: a row that
79 // says unsubscribed with no unsubscribed_at cannot answer "when", which is
80 // the first question asked of an opt-out.
81 let unsubscribed_at = (state == SubscriptionState::Unsubscribed).then(chrono::Utc::now);
82
83 let mut tx = pool.begin().await?;
84
85 // The uniqueness of a subscriber is enforced by two partial indexes, one
86 // per identity kind, and ON CONFLICT has to name the one that applies.
87 // A single statement cannot cover both, so the arm is chosen here rather
88 // than left to the database to guess.
89 let conflict_target = match subscriber {
90 Subscriber::User(_) => "(list_id, user_id) WHERE user_id IS NOT NULL",
91 Subscriber::Email(_) => "(list_id, email) WHERE email IS NOT NULL",
92 };
93 let sql = format!(
94 "INSERT INTO list_subscriptions \
95 (list_id, user_id, email, state, source, confirmed_at, unsubscribed_at) \
96 VALUES ($1, $2, $3, $4, $5, $6, $7) \
97 ON CONFLICT {conflict_target} \
98 DO UPDATE SET state = EXCLUDED.state, \
99 confirmed_at = EXCLUDED.confirmed_at, \
100 unsubscribed_at = EXCLUDED.unsubscribed_at \
101 RETURNING id"
102 );
103
104 let subscription_id = sqlx::query_scalar::<_, ListSubscriptionId>(&sql)
105 .bind(list_id)
106 .bind(user_id)
107 .bind(email.as_deref())
108 .bind(state.to_string())
109 .bind(source.to_string())
110 .bind(confirmed_at)
111 .bind(unsubscribed_at)
112 .fetch_one(&mut *tx)
113 .await?;
114
115 sqlx::query(
116 "INSERT INTO consent_events (subscription_id, event, evidence) VALUES ($1, $2, $3)",
117 )
118 .bind(subscription_id)
119 .bind(event.to_string())
120 .bind(evidence)
121 .execute(&mut *tx)
122 .await?;
123
124 tx.commit().await?;
125 Ok(subscription_id)
126 }
127
128 /// Mark a subscription unsubscribed and append the opt-out event.
129 ///
130 /// Returns whether a subscription moved. Idempotent: unsubscribing twice
131 /// reports `false` the second time and is not an error, which matters because
132 /// RFC 8058 one-click POSTs get retried.
133 #[tracing::instrument(skip_all)]
134 pub async fn unsubscribe(
135 pool: &PgPool,
136 subscription_id: ListSubscriptionId,
137 event: ConsentEvent,
138 ) -> Result<bool> {
139 let mut tx = pool.begin().await?;
140
141 let moved = sqlx::query(
142 "UPDATE list_subscriptions SET state = 'unsubscribed', unsubscribed_at = NOW() \
143 WHERE id = $1 AND state <> 'unsubscribed'",
144 )
145 .bind(subscription_id)
146 .execute(&mut *tx)
147 .await?
148 .rows_affected()
149 > 0;
150
151 if moved {
152 sqlx::query("INSERT INTO consent_events (subscription_id, event) VALUES ($1, $2)")
153 .bind(subscription_id)
154 .bind(event.to_string())
155 .execute(&mut *tx)
156 .await?;
157 }
158
159 tx.commit().await?;
160 Ok(moved)
161 }
162
163 /// The states that may receive mail.
164 ///
165 /// `imported` is here to preserve pre-migration behaviour, not because we hold
166 /// evidence of consent for those rows. Before the unified tables, everyone in
167 /// `mailing_list_subscribers` was mailed, and a refactor whose side effect is
168 /// that some subscribers silently stop receiving mail is worse than one that
169 /// changes nothing.
170 ///
171 /// Imported subscribers stay sendable, marketing included, and there is no
172 /// re-confirmation pass. The existing signups are treated as valid consent.
173 /// That is a deliberate acceptance of the fact that nothing recorded what any
174 /// given subscriber was told they were signing up for, taken against the cost
175 /// of losing most of the existing list.
176 ///
177 /// `Imported` therefore survives as provenance only: it says how the row got
178 /// here, not whether the row may be mailed. Nothing gates on it.
179 const SENDABLE_STATES: &[&str] = &["confirmed", "imported"];
180
181 /// One deliverable recipient.
182 #[derive(Debug, Clone, sqlx::FromRow)]
183 pub struct Recipient {
184 /// The subscription this delivery is against. Carried so the caller can
185 /// mint a per-recipient unsubscribe link and, later, record the send.
186 pub subscription_id: ListSubscriptionId,
187 /// `None` for a bare address with no account behind it.
188 pub user_id: Option<UserId>,
189 pub email: String,
190 pub display_name: Option<String>,
191 }
192
193 /// A list and everyone who may currently be mailed on it.
194 #[derive(Debug, Clone)]
195 pub struct Audience {
196 pub list_id: ListId,
197 /// Transactional list nobody may leave, so no unsubscribe footer is owed.
198 /// The caller reads this rather than deciding per send, which is what stops
199 /// a marketing send from quietly omitting the footer.
200 pub required: bool,
201 pub recipients: Vec<Recipient>,
202 }
203
204 /// Everyone who may be mailed on a list, and nobody who may not.
205 ///
206 /// This is the one place the delivery rules live. They were previously spread
207 /// across each send's own query, which is why suppression was applied
208 /// consistently (it sat in `send_email_inner`) and nothing else was.
209 ///
210 /// Applied here, in order:
211 /// - the subscription state must be sendable (see [`SENDABLE_STATES`]);
212 /// - the address must not be suppressed, which covers bounces and complaints;
213 /// - an account subscriber must have a verified, unsuspended account. A bare
214 /// address has no account to check, and excluding those was a real bug once:
215 /// an INNER JOIN meant imported subscribers were never mailed.
216 ///
217 /// Capped at 10,000, matching the query it replaces.
218 #[tracing::instrument(skip_all)]
219 pub async fn resolve_audience(pool: &PgPool, list_id: ListId) -> Result<Audience> {
220 let required = sqlx::query_scalar::<_, bool>("SELECT required FROM lists WHERE id = $1")
221 .bind(list_id)
222 .fetch_one(pool)
223 .await?;
224
225 let recipients = sqlx::query_as::<_, Recipient>(
226 r"
227 SELECT ls.id AS subscription_id, u.id AS user_id, u.email, u.display_name
228 FROM list_subscriptions ls
229 JOIN users u ON u.id = ls.user_id
230 WHERE ls.list_id = $1
231 AND ls.state = ANY($2)
232 AND u.email_verified = true
233 AND u.suspended_at IS NULL
234 AND LOWER(u.email) NOT IN (SELECT LOWER(email) FROM email_suppressions)
235 UNION ALL
236 SELECT ls.id AS subscription_id, NULL::uuid AS user_id, ls.email, NULL AS display_name
237 FROM list_subscriptions ls
238 WHERE ls.list_id = $1
239 AND ls.state = ANY($2)
240 AND ls.user_id IS NULL
241 AND ls.email IS NOT NULL
242 AND LOWER(ls.email) NOT IN (SELECT LOWER(email) FROM email_suppressions)
243 LIMIT 10000
244 ",
245 )
246 .bind(list_id)
247 .bind(SENDABLE_STATES)
248 .fetch_all(pool)
249 .await?;
250
251 Ok(Audience {
252 list_id,
253 required,
254 recipients,
255 })
256 }
257
258 /// The unified list mirroring a legacy per-project list.
259 ///
260 /// Resolves through `mailing_lists` rather than storing a foreign key, so the
261 /// old table needs no schema change during the migration and dropping it later
262 /// leaves nothing dangling.
263 #[tracing::instrument(skip_all)]
264 pub async fn list_for_legacy(pool: &PgPool, mailing_list_id: uuid::Uuid) -> Result<Option<ListId>> {
265 let id = sqlx::query_scalar::<_, ListId>(
266 "SELECT l.id FROM mailing_lists ml \
267 JOIN lists l ON l.scope = 'project' AND l.scope_id = ml.project_id AND l.kind = ml.list_type \
268 WHERE ml.id = $1",
269 )
270 .bind(mailing_list_id)
271 .fetch_optional(pool)
272 .await?;
273 Ok(id)
274 }
275
276 /// Count subscriptions on a list in a given state. Exists for the backfill
277 /// tests and the admin view; the send path uses [`resolve_audience`].
278 #[tracing::instrument(skip_all)]
279 pub async fn count_in_state(
280 pool: &PgPool,
281 list_id: ListId,
282 state: SubscriptionState,
283 ) -> Result<i64> {
284 let count = sqlx::query_scalar::<_, i64>(
285 "SELECT COUNT(*) FROM list_subscriptions WHERE list_id = $1 AND state = $2",
286 )
287 .bind(list_id)
288 .bind(state.to_string())
289 .fetch_one(pool)
290 .await?;
291 Ok(count)
292 }
293
294 /// One row on the unsubscribe page.
295 #[derive(Debug, Clone, sqlx::FromRow)]
296 pub struct SubscriptionRow {
297 pub subscription_id: ListSubscriptionId,
298 pub list_id: ListId,
299 pub title: String,
300 /// Transactional list. Shown so the page is an honest inventory of what we
301 /// send, but it carries no toggle: there is no opting out of a receipt.
302 pub required: bool,
303 pub state: String,
304 /// The creator who owns the list, if one does.
305 ///
306 /// `None` for a platform list (receipts, security, account mail), which
307 /// nobody but us decides. Present for a project list, where the creator is
308 /// the other joint controller of the address and the preferences page has
309 /// to be able to say so. See `site-docs/public/legal/mailing-list-data-processing.md`.
310 pub owner_name: Option<String>,
311 }
312
313 impl SubscriptionRow {
314 /// Whether this row is currently receiving mail.
315 pub fn subscribed(&self) -> bool {
316 SENDABLE_STATES.contains(&self.state.as_str())
317 }
318 }
319
320 /// Every list the subscriber behind `subscription_id` is on.
321 ///
322 /// The page is reached from a link in one email, but it shows all of them:
323 /// somebody who wants out is rarely asking about the single list that happened
324 /// to prompt them, and making them hunt for the rest is how "unsubscribe" turns
325 /// into "mark as spam".
326 #[tracing::instrument(skip_all)]
327 pub async fn subscriptions_for_peer(
328 pool: &PgPool,
329 subscription_id: ListSubscriptionId,
330 ) -> Result<Vec<SubscriptionRow>> {
331 let rows = sqlx::query_as::<_, SubscriptionRow>(
332 r"
333 WITH peer AS (
334 SELECT user_id, email FROM list_subscriptions WHERE id = $1
335 )
336 SELECT ls.id AS subscription_id, l.id AS list_id, l.title, l.required, ls.state,
337 COALESCE(NULLIF(u.display_name, ''), u.username) AS owner_name
338 FROM list_subscriptions ls
339 JOIN lists l ON l.id = ls.list_id
340 LEFT JOIN users u ON u.id = l.owner_id
341 CROSS JOIN peer
342 WHERE (peer.user_id IS NOT NULL AND ls.user_id = peer.user_id)
343 OR (peer.email IS NOT NULL AND LOWER(ls.email) = LOWER(peer.email))
344 ORDER BY l.required DESC, l.title
345 ",
346 )
347 .bind(subscription_id)
348 .fetch_all(pool)
349 .await?;
350 Ok(rows)
351 }
352
353 /// Unsubscribe the peer from every list they may leave.
354 ///
355 /// Required lists are skipped rather than refused: "unsubscribe from
356 /// everything" means everything on offer, and a receipt was never on offer.
357 /// Returns how many moved.
358 #[tracing::instrument(skip_all)]
359 pub async fn unsubscribe_peer_from_all(
360 pool: &PgPool,
361 subscription_id: ListSubscriptionId,
362 ) -> Result<usize> {
363 let rows = subscriptions_for_peer(pool, subscription_id).await?;
364 let mut moved = 0;
365 for row in rows {
366 if row.required {
367 continue;
368 }
369 if unsubscribe(pool, row.subscription_id, ConsentEvent::OptOut).await? {
370 moved += 1;
371 }
372 }
373 Ok(moved)
374 }
375
376 /// Re-subscribe a row the page had toggled off. Appends an `opt_in`; the
377 /// `opt_out` before it stays, because the history is the point.
378 #[tracing::instrument(skip_all)]
379 pub async fn resubscribe(pool: &PgPool, subscription_id: ListSubscriptionId) -> Result<bool> {
380 let mut tx = pool.begin().await?;
381 let moved = sqlx::query(
382 "UPDATE list_subscriptions \
383 SET state = 'confirmed', confirmed_at = NOW(), unsubscribed_at = NULL \
384 WHERE id = $1 AND state = 'unsubscribed'",
385 )
386 .bind(subscription_id)
387 .execute(&mut *tx)
388 .await?
389 .rows_affected()
390 > 0;
391
392 if moved {
393 sqlx::query(
394 "INSERT INTO consent_events (subscription_id, event, evidence) VALUES ($1, 'opt_in', $2)",
395 )
396 .bind(subscription_id)
397 .bind("Re-subscribed from the email preferences page.")
398 .execute(&mut *tx)
399 .await?;
400 }
401 tx.commit().await?;
402 Ok(moved)
403 }
404
405 /// Whether a subscription's list may be left at all.
406 #[tracing::instrument(skip_all)]
407 pub async fn subscription_is_required(
408 pool: &PgPool,
409 subscription_id: ListSubscriptionId,
410 ) -> Result<bool> {
411 let required = sqlx::query_scalar::<_, bool>(
412 "SELECT l.required FROM list_subscriptions ls \
413 JOIN lists l ON l.id = ls.list_id WHERE ls.id = $1",
414 )
415 .bind(subscription_id)
416 .fetch_optional(pool)
417 .await?;
418 Ok(required.unwrap_or(false))
419 }
420
421 // ── Per-repo notification lists ──
422
423 /// Whether this user has opted out of a repo's notifications.
424 ///
425 /// The gate is "not opted out" rather than "opted in", which is what keeps the
426 /// behaviour identical to the account-wide bool it replaces: eligibility is
427 /// still repo ownership or issue participation, and an absent row still means
428 /// nothing has been said. Opting in to a repo you have nothing to do with would
429 /// not get you mail, because it would not make you a participant.
430 #[tracing::instrument(skip_all)]
431 pub async fn repo_notifications_muted(
432 pool: &PgPool,
433 repo_id: uuid::Uuid,
434 user_id: UserId,
435 kind: ListKind,
436 ) -> Result<bool> {
437 let muted = sqlx::query_scalar::<_, bool>(
438 "SELECT EXISTS( \
439 SELECT 1 FROM list_subscriptions ls \
440 JOIN lists l ON l.id = ls.list_id \
441 WHERE l.scope = 'repo' AND l.scope_id = $1 AND l.kind = $2 \
442 AND ls.user_id = $3 AND ls.state = 'unsubscribed')",
443 )
444 .bind(repo_id)
445 .bind(kind.to_string())
446 .bind(user_id)
447 .fetch_one(pool)
448 .await?;
449 Ok(muted)
450 }
451
452 /// Mute or unmute a repo's notifications for one user.
453 ///
454 /// Muting records an explicit `unsubscribed` row; unmuting moves it back.
455 /// Either way a consent event is appended, so "when did I turn this off" has an
456 /// answer.
457 #[tracing::instrument(skip_all)]
458 pub async fn set_repo_muted(
459 pool: &PgPool,
460 repo_id: uuid::Uuid,
461 user_id: UserId,
462 kind: ListKind,
463 muted: bool,
464 ) -> Result<()> {
465 let Some(list_id) = find_list(pool, ListScope::Repo, Some(repo_id), kind).await? else {
466 return Ok(());
467 };
468
469 if muted {
470 subscribe(
471 pool,
472 list_id,
473 &Subscriber::User(user_id),
474 SubscriptionState::Unsubscribed,
475 SubscriptionSource::ProjectPage,
476 ConsentEvent::OptOut,
477 Some("Muted from the repository page."),
478 )
479 .await?;
480 } else {
481 subscribe(
482 pool,
483 list_id,
484 &Subscriber::User(user_id),
485 SubscriptionState::Confirmed,
486 SubscriptionSource::ProjectPage,
487 ConsentEvent::OptIn,
488 Some("Unmuted from the repository page."),
489 )
490 .await?;
491 }
492 Ok(())
493 }
494
495 // ── Account notification preferences ──
496 //
497 // Subscriptions are the whole record. Seven bool columns on `users` held these
498 // until migration 189; every read moved in 5b and the columns went in 5c, so
499 // there is one place a preference lives and one place it is read from.
500 //
501 // What is left of the columns is their names, in NOTIFICATION_LISTS below:
502 // unsubscribe links already sent carry them in signed URLs.
503
504 /// Platform notification lists, paired with the legacy preference name.
505 ///
506 /// The second element was the `users.notify_*` column until migration 189
507 /// dropped them. It survives because unsubscribe links already sitting in
508 /// inboxes carry those names in their signed URLs, and renaming them would
509 /// invalidate every link ever sent. `disable_notification` maps them back.
510 pub const NOTIFICATION_LISTS: &[(&str, &str)] = &[
511 ("sale", "notify_sale"),
512 ("follower", "notify_follower"),
513 ("releases", "notify_release"),
514 ("issues", "notify_issues"),
515 ("status", "notify_status"),
516 ("tip", "notify_tip"),
517 ("login", "login_notification_enabled"),
518 ];
519
520 /// The `users` column a platform list mirrors, if it mirrors one.
521 pub fn notification_column_for_kind(kind: &str) -> Option<&'static str> {
522 NOTIFICATION_LISTS
523 .iter()
524 .find(|(k, _)| *k == kind)
525 .map(|(_, col)| *col)
526 }
527
528 /// What a preference is when nobody has said otherwise.
529 ///
530 /// Matches the `users` column defaults, which is where these lived until the
531 /// reads moved. Only reached if a subscription row is missing, which the 186
532 /// backfill and the 187 trigger between them should make impossible; the
533 /// fallback exists so a missing row degrades to the documented default rather
534 /// than to silence. `notification_rows_exist_for_every_account` is the test
535 /// that keeps it unreachable.
536 fn default_enabled(kind: &str) -> bool {
537 // Status alerts are the one opt-in: they are platform operations noise, and
538 // a new account has not asked for them. Everything else, `invite` included
539 // (migration 195, which has no column behind it), is on until turned off.
540 kind != "status"
541 }
542
543 /// May this user be sent this kind of notification?
544 ///
545 /// The single read for every account notification. Each of these used to be a
546 /// `users.notify_*` column consulted at the send site, which is why the rules
547 /// could differ per site and why opting out was all-or-nothing.
548 ///
549 /// A missing subscription falls back to [`default_enabled`] rather than
550 /// refusing: the failure mode of a bug here should be mail somebody expected,
551 /// not silence they cannot diagnose.
552 #[tracing::instrument(skip_all)]
553 pub async fn may_notify(pool: &PgPool, user_id: UserId, kind: ListKind) -> Result<bool> {
554 let kind = kind.to_string();
555 let state = sqlx::query_scalar::<_, String>(
556 "SELECT ls.state FROM list_subscriptions ls \
557 JOIN lists l ON l.id = ls.list_id \
558 WHERE l.scope = 'platform' AND l.kind = $1 AND ls.user_id = $2",
559 )
560 .bind(&kind)
561 .bind(user_id)
562 .fetch_optional(pool)
563 .await?;
564
565 Ok(match state {
566 Some(s) => SENDABLE_STATES.contains(&s.as_str()),
567 None => {
568 tracing::warn!(
569 user_id = %user_id, kind = %kind,
570 "no notification subscription row; falling back to the default"
571 );
572 default_enabled(&kind)
573 }
574 })
575 }
576
577 /// Every notification preference for one user, for the settings screen.
578 ///
579 /// One query rather than seven `may_notify` calls, because the account tab
580 /// renders all of them at once. Missing rows fall back to the same defaults
581 /// `may_notify` uses, so the two cannot disagree about an account the trigger
582 /// somehow missed.
583 #[derive(Debug, Clone)]
584 pub struct NotificationPrefs {
585 pub sale: bool,
586 pub follower: bool,
587 pub release: bool,
588 pub login: bool,
589 pub issues: bool,
590 pub status: bool,
591 pub tip: bool,
592 pub invite: bool,
593 }
594
595 #[tracing::instrument(skip_all)]
596 pub async fn notification_prefs(pool: &PgPool, user_id: UserId) -> Result<NotificationPrefs> {
597 let rows = sqlx::query_as::<_, (String, String)>(
598 "SELECT l.kind, ls.state FROM list_subscriptions ls \
599 JOIN lists l ON l.id = ls.list_id \
600 WHERE l.scope = 'platform' AND ls.user_id = $1",
601 )
602 .bind(user_id)
603 .fetch_all(pool)
604 .await?;
605
606 let enabled = |kind: &str| {
607 rows.iter().find(|(k, _)| k == kind).map_or_else(
608 || default_enabled(kind),
609 |(_, state)| SENDABLE_STATES.contains(&state.as_str()),
610 )
611 };
612
613 Ok(NotificationPrefs {
614 sale: enabled("sale"),
615 follower: enabled("follower"),
616 release: enabled("releases"),
617 login: enabled("login"),
618 issues: enabled("issues"),
619 status: enabled("status"),
620 tip: enabled("tip"),
621 invite: enabled("invite"),
622 })
623 }
624
625 /// Point a user's notification subscription at `enabled`, appending the consent
626 /// event that goes with it. Called after the column is written.
627 #[tracing::instrument(skip_all)]
628 pub async fn sync_notification_subscription(
629 pool: &PgPool,
630 user_id: UserId,
631 kind: &str,
632 enabled: bool,
633 ) -> Result<()> {
634 let Some(list_id) = sqlx::query_scalar::<_, ListId>(
635 "SELECT id FROM lists WHERE scope = 'platform' AND kind = $1",
636 )
637 .bind(kind)
638 .fetch_optional(pool)
639 .await?
640 else {
641 return Ok(());
642 };
643
644 if enabled {
645 subscribe(
646 pool,
647 list_id,
648 &Subscriber::User(user_id),
649 SubscriptionState::Confirmed,
650 SubscriptionSource::Admin,
651 ConsentEvent::OptIn,
652 Some("Enabled from account notification settings."),
653 )
654 .await?;
655 return Ok(());
656 }
657
658 let existing = sqlx::query_scalar::<_, ListSubscriptionId>(
659 "SELECT id FROM list_subscriptions WHERE list_id = $1 AND user_id = $2",
660 )
661 .bind(list_id)
662 .bind(user_id)
663 .fetch_optional(pool)
664 .await?;
665
666 match existing {
667 Some(subscription_id) => {
668 unsubscribe(pool, subscription_id, ConsentEvent::OptOut).await?;
669 }
670 None => {
671 // No row yet (an account created since the backfill). Record the
672 // "no" rather than leaving it absent, so it reads as a choice
673 // rather than as never having been asked.
674 subscribe(
675 pool,
676 list_id,
677 &Subscriber::User(user_id),
678 SubscriptionState::Unsubscribed,
679 SubscriptionSource::Admin,
680 ConsentEvent::OptOut,
681 Some("Disabled from account notification settings."),
682 )
683 .await?;
684 }
685 }
686 Ok(())
687 }
688
689 /// The `users` column behind a subscription, if it has one.
690 ///
691 /// Used by the preferences page: a toggle there has to reach the column, or the
692 /// send that reads the column will ignore it.
693 #[tracing::instrument(skip_all)]
694 pub async fn notification_column_for_subscription(
695 pool: &PgPool,
696 subscription_id: ListSubscriptionId,
697 ) -> Result<Option<(UserId, &'static str)>> {
698 let row = sqlx::query_as::<_, (Option<UserId>, String, String)>(
699 "SELECT ls.user_id, l.scope, l.kind FROM list_subscriptions ls \
700 JOIN lists l ON l.id = ls.list_id WHERE ls.id = $1",
701 )
702 .bind(subscription_id)
703 .fetch_optional(pool)
704 .await?;
705
706 let Some((Some(user_id), scope, kind)) = row else {
707 return Ok(None);
708 };
709 if scope != "platform" {
710 return Ok(None);
711 }
712 Ok(notification_column_for_kind(&kind).map(|col| (user_id, col)))
713 }
714
715 // ── Mirroring the legacy tables ──
716 //
717 // `mailing_lists` / `mailing_list_subscribers` are still what the product
718 // writes to, and `resolve_audience` is what sends now read. Every legacy write
719 // therefore has to reach here, or a subscriber added after the migration is one
720 // no send can see.
721 //
722 // These propagate their errors rather than logging and continuing. A subscribe
723 // that does not reach the send path is a broken subscribe, and the failure
724 // should be visible where it happened rather than at the next announcement.
725
726 /// Mirror a legacy project list into `lists`.
727 #[tracing::instrument(skip_all)]
728 pub async fn mirror_legacy_list(
729 pool: &PgPool,
730 project_id: uuid::Uuid,
731 kind: ListKind,
732 title: &str,
733 ) -> Result<()> {
734 sqlx::query(
735 "INSERT INTO lists (scope, scope_id, kind, title, required, owner_id) \
736 SELECT 'project', $1, $2, $3, FALSE, p.user_id FROM projects p WHERE p.id = $1 \
737 ON CONFLICT DO NOTHING",
738 )
739 .bind(project_id)
740 .bind(kind.to_string())
741 .bind(title)
742 .execute(pool)
743 .await?;
744 Ok(())
745 }
746
747 /// Mirror a legacy subscribe.
748 ///
749 /// A subscribe through the product is a real act, so it lands `confirmed` with
750 /// an `opt_in` event, unlike the backfill's `imported`. `evidence` records what
751 /// the person was doing at the time, which is the difference between consent we
752 /// can show and consent we assert.
753 #[tracing::instrument(skip_all)]
754 pub async fn mirror_legacy_subscribe(
755 pool: &PgPool,
756 mailing_list_id: uuid::Uuid,
757 subscriber: &Subscriber,
758 state: SubscriptionState,
759 source: SubscriptionSource,
760 evidence: Option<&str>,
761 ) -> Result<()> {
762 let Some(list_id) = list_for_legacy(pool, mailing_list_id).await? else {
763 // The legacy list predates its mirror. Create-list mirroring runs first
764 // for anything made since the migration, so this means a list that was
765 // never backfilled, which is a bug worth seeing rather than skipping.
766 return Err(crate::error::AppError::Internal(anyhow::anyhow!(
767 "legacy mailing list {mailing_list_id} has no unified list"
768 )));
769 };
770 let event = match state {
771 SubscriptionState::Imported => ConsentEvent::Import,
772 _ => ConsentEvent::OptIn,
773 };
774 subscribe(pool, list_id, subscriber, state, source, event, evidence).await?;
775 Ok(())
776 }
777
778 /// Mirror a legacy unsubscribe for an account subscriber.
779 ///
780 /// The most important mirror of the three: a missed unsubscribe means mailing
781 /// somebody who asked us not to.
782 #[tracing::instrument(skip_all)]
783 pub async fn mirror_legacy_unsubscribe_user(
784 pool: &PgPool,
785 mailing_list_id: uuid::Uuid,
786 user_id: UserId,
787 ) -> Result<()> {
788 let Some(list_id) = list_for_legacy(pool, mailing_list_id).await? else {
789 return Ok(());
790 };
791 mark_unsubscribed(pool, list_id, &Subscriber::User(user_id)).await
792 }
793
794 /// Mirror a legacy unsubscribe for a bare address.
795 #[tracing::instrument(skip_all)]
796 pub async fn mirror_legacy_unsubscribe_email(
797 pool: &PgPool,
798 mailing_list_id: uuid::Uuid,
799 email: &str,
800 ) -> Result<()> {
801 let Some(list_id) = list_for_legacy(pool, mailing_list_id).await? else {
802 return Ok(());
803 };
804 mark_unsubscribed(pool, list_id, &Subscriber::Email(email.to_string())).await
805 }
806
807 /// Mirror an unsubscribe from every list on a project (the unfollow path).
808 #[tracing::instrument(skip_all)]
809 pub async fn mirror_legacy_unsubscribe_project(
810 pool: &PgPool,
811 project_id: uuid::Uuid,
812 user_id: UserId,
813 ) -> Result<()> {
814 let ids = sqlx::query_scalar::<_, ListId>(
815 "SELECT id FROM lists WHERE scope = 'project' AND scope_id = $1",
816 )
817 .bind(project_id)
818 .fetch_all(pool)
819 .await?;
820 for list_id in ids {
821 mark_unsubscribed(pool, list_id, &Subscriber::User(user_id)).await?;
822 }
823 Ok(())
824 }
825
826 /// Move a subscription to `unsubscribed` and append the opt-out, by identity
827 /// rather than by subscription id. No-op when there is nothing subscribed.
828 async fn mark_unsubscribed(pool: &PgPool, list_id: ListId, subscriber: &Subscriber) -> Result<()> {
829 let existing =
830 match subscriber {
831 Subscriber::User(id) => {
832 sqlx::query_scalar::<_, ListSubscriptionId>(
833 "SELECT id FROM list_subscriptions WHERE list_id = $1 AND user_id = $2",
834 )
835 .bind(list_id)
836 .bind(id)
837 .fetch_optional(pool)
838 .await?
839 }
840 Subscriber::Email(addr) => sqlx::query_scalar::<_, ListSubscriptionId>(
841 "SELECT id FROM list_subscriptions WHERE list_id = $1 AND LOWER(email) = LOWER($2)",
842 )
843 .bind(list_id)
844 .bind(addr)
845 .fetch_optional(pool)
846 .await?,
847 };
848 if let Some(subscription_id) = existing {
849 unsubscribe(pool, subscription_id, ConsentEvent::OptOut).await?;
850 }
851 Ok(())
852 }
853
854 #[cfg(test)]
855 mod tests {
856 use super::*;
857
858 #[test]
859 fn scope_and_kind_round_trip() {
860 for s in [
861 ListScope::Platform,
862 ListScope::Project,
863 ListScope::Repo,
864 ListScope::Creator,
865 ] {
866 assert_eq!(s.to_string().parse::<ListScope>().unwrap(), s);
867 }
868 for k in [ListKind::Content, ListKind::Marketing, ListKind::Issues] {
869 assert_eq!(k.to_string().parse::<ListKind>().unwrap(), k);
870 }
871 }
872
873 /// Every `ListKind` is accepted by the database.
874 ///
875 /// A kind lives in two places: this enum and the `lists_kind_check`
876 /// constraint. Adding it to only the enum compiles, passes every unit test,
877 /// and then fails at INSERT against a deployed database, which is a long
878 /// way from the edit that caused it. So the constraint is read back here.
879 ///
880 /// Reads the last migration that redefines the constraint, since each one
881 /// replaces the previous in full (186, then 195).
882 #[test]
883 fn every_kind_is_allowed_by_the_check_constraint() {
884 let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
885 let mut files: Vec<_> = std::fs::read_dir(&dir)
886 .expect("migrations directory")
887 .filter_map(|e| e.ok().map(|e| e.path()))
888 .filter(|p| {
889 std::fs::read_to_string(p).is_ok_and(|s| s.contains("lists_kind_check CHECK"))
890 })
891 .collect();
892 files.sort();
893 let newest = files
894 .last()
895 .expect("some migration defines lists_kind_check");
896 let sql = std::fs::read_to_string(newest).expect("readable migration");
897
898 let clause = sql
899 .split_once("lists_kind_check CHECK (kind IN (")
900 .expect("the constraint has the expected shape")
901 .1
902 .split_once("))")
903 .expect("the constraint list is closed")
904 .0;
905 let allowed: Vec<&str> = clause
906 .split(',')
907 .map(|s| s.trim().trim_matches('\'').trim())
908 .filter(|s| !s.is_empty())
909 .collect();
910
911 for kind in ListKind::ALL {
912 let s = kind.to_string();
913 assert!(
914 allowed.contains(&s.as_str()),
915 "ListKind::{kind:?} (\"{s}\") is not in the lists_kind_check constraint. \
916 Adding a kind takes a migration as well as an enum variant, or the first \
917 insert of one fails on a deployed database.",
918 );
919 }
920 assert_eq!(
921 allowed.len(),
922 ListKind::ALL.len(),
923 "the constraint allows {allowed:?}, which is not the set ListKind names. A kind \
924 the database accepts but the enum cannot represent is unreachable from the code.",
925 );
926 }
927
928 /// Which states receive mail is a policy, and a settled one. Changing this
929 /// set changes who gets email, so it should be an edit somebody made on
930 /// purpose rather than a line that moved during a refactor.
931 #[test]
932 fn sendable_states_are_the_agreed_set() {
933 assert_eq!(
934 SENDABLE_STATES,
935 &["confirmed", "imported"],
936 "'imported' is sendable by decision (GoingsOn 04a882b4, 2026-08-06): no \
937 re-confirmation pass, the existing signups count as consent. Dropping it \
938 silently stops mail for most of the list, so it needs a new decision and \
939 not just a diff"
940 );
941 }
942
943 /// The strings are a database CHECK constraint, so a rename here that is
944 /// not matched by a migration fails at insert rather than at compile time.
945 #[test]
946 fn state_and_event_strings_match_the_check_constraints() {
947 assert_eq!(SubscriptionState::Imported.to_string(), "imported");
948 assert_eq!(SubscriptionState::Unsubscribed.to_string(), "unsubscribed");
949 assert_eq!(SubscriptionSource::LandingForm.to_string(), "landing_form");
950 assert_eq!(ConsentEvent::AdminRemoval.to_string(), "admin_removal");
951 assert_eq!(ConsentEvent::Import.to_string(), "import");
952 }
953 }
954