Skip to main content

max / makenotwork

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