//! One subscription model for every list-like thing.
//!
//! Step 2 of the plan in the maintainer wiki (`mnw-mailing-lists`). Three
//! tables: a `list` is anything somebody can be subscribed to, a
//! `list_subscription` is one recipient's relationship to one list, and a
//! `consent_event` is why we believe we are allowed to mail them.
//!
//! **Nothing sends through this yet.** `mailing_lists` and `email_signups`
//! remain authoritative until the resolver lands (step 3). This module exists
//! so the schema has typed access and the backfill has tests, not so callers
//! can start using it piecemeal, which is how the five parallel mechanisms this
//! replaces came to exist in the first place.
//! <!-- wiki: mnw-mailing-lists -->

use sqlx::PgPool;

use super::enums::{ConsentEvent, ListKind, ListScope, SubscriptionSource, SubscriptionState};
use super::id_types::{ListId, ListSubscriptionId, UserId};
use crate::error::Result;

/// A subscriber: an account or a bare address, never both. The old
/// `mailing_list_subscribers` allowed both at once and left "which one is
/// authoritative" to whoever read the row next; the new table's CHECK makes
/// that unrepresentable, and this type carries the same rule into Rust.
#[derive(Debug, Clone)]
pub enum Subscriber {
    User(UserId),
    Email(String),
}

/// Find a list by what it is attached to.
///
/// `scope_id` must be `None` for [`ListScope::Platform`] and `Some` otherwise;
/// the database CHECK rejects the other combinations, so a mismatch here
/// returns no row rather than the wrong one.
#[tracing::instrument(skip_all)]
pub async fn find_list(
    pool: &PgPool,
    scope: ListScope,
    scope_id: Option<uuid::Uuid>,
    kind: ListKind,
) -> Result<Option<ListId>> {
    let id = sqlx::query_scalar::<_, ListId>(
        "SELECT id FROM lists \
         WHERE scope = $1 AND kind = $2 \
           AND (scope_id = $3 OR ($3::uuid IS NULL AND scope_id IS NULL))",
    )
    .bind(scope.to_string())
    .bind(kind.to_string())
    .bind(scope_id)
    .fetch_optional(pool)
    .await?;
    Ok(id)
}

/// Record a subscription and the consent event that justifies it, in one
/// transaction.
///
/// The two are written together because a subscription without its consent
/// event is exactly the state this whole model exists to eliminate: someone on
/// a list with no record of why. Re-subscribing an address that had left moves
/// it back and appends a fresh event rather than editing the old one.
#[tracing::instrument(skip_all)]
pub async fn subscribe(
    pool: &PgPool,
    list_id: ListId,
    subscriber: &Subscriber,
    state: SubscriptionState,
    source: SubscriptionSource,
    event: ConsentEvent,
    evidence: Option<&str>,
) -> Result<ListSubscriptionId> {
    let (user_id, email) = match subscriber {
        Subscriber::User(id) => (Some(*id), None),
        Subscriber::Email(addr) => (None, Some(addr.to_lowercase())),
    };
    let confirmed_at = (state == SubscriptionState::Confirmed).then(chrono::Utc::now);
    // Kept in step with `state` here rather than at each call site: a row that
    // says unsubscribed with no unsubscribed_at cannot answer "when", which is
    // the first question asked of an opt-out.
    let unsubscribed_at = (state == SubscriptionState::Unsubscribed).then(chrono::Utc::now);

    let mut tx = pool.begin().await?;

    // The uniqueness of a subscriber is enforced by two partial indexes, one
    // per identity kind, and ON CONFLICT has to name the one that applies.
    // A single statement cannot cover both, so the arm is chosen here rather
    // than left to the database to guess.
    let conflict_target = match subscriber {
        Subscriber::User(_) => "(list_id, user_id) WHERE user_id IS NOT NULL",
        Subscriber::Email(_) => "(list_id, email) WHERE email IS NOT NULL",
    };
    let sql = format!(
        "INSERT INTO list_subscriptions \
             (list_id, user_id, email, state, source, confirmed_at, unsubscribed_at) \
         VALUES ($1, $2, $3, $4, $5, $6, $7) \
         ON CONFLICT {conflict_target} \
         DO UPDATE SET state = EXCLUDED.state, \
                       confirmed_at = EXCLUDED.confirmed_at, \
                       unsubscribed_at = EXCLUDED.unsubscribed_at \
         RETURNING id"
    );

    let subscription_id = sqlx::query_scalar::<_, ListSubscriptionId>(&sql)
        .bind(list_id)
        .bind(user_id)
        .bind(email.as_deref())
        .bind(state.to_string())
        .bind(source.to_string())
        .bind(confirmed_at)
        .bind(unsubscribed_at)
        .fetch_one(&mut *tx)
        .await?;

    sqlx::query(
        "INSERT INTO consent_events (subscription_id, event, evidence) VALUES ($1, $2, $3)",
    )
    .bind(subscription_id)
    .bind(event.to_string())
    .bind(evidence)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(subscription_id)
}

/// Mark a subscription unsubscribed and append the opt-out event.
///
/// Returns whether a subscription moved. Idempotent: unsubscribing twice
/// reports `false` the second time and is not an error, which matters because
/// RFC 8058 one-click POSTs get retried.
#[tracing::instrument(skip_all)]
pub async fn unsubscribe(
    pool: &PgPool,
    subscription_id: ListSubscriptionId,
    event: ConsentEvent,
) -> Result<bool> {
    let mut tx = pool.begin().await?;

    let moved = sqlx::query(
        "UPDATE list_subscriptions SET state = 'unsubscribed', unsubscribed_at = NOW() \
         WHERE id = $1 AND state <> 'unsubscribed'",
    )
    .bind(subscription_id)
    .execute(&mut *tx)
    .await?
    .rows_affected()
        > 0;

    if moved {
        sqlx::query("INSERT INTO consent_events (subscription_id, event) VALUES ($1, $2)")
            .bind(subscription_id)
            .bind(event.to_string())
            .execute(&mut *tx)
            .await?;
    }

    tx.commit().await?;
    Ok(moved)
}

/// The states that may receive mail.
///
/// `imported` is here to preserve pre-migration behaviour, not because we hold
/// evidence of consent for those rows. Before the unified tables, everyone in
/// `mailing_list_subscribers` was mailed, and a refactor whose side effect is
/// that some subscribers silently stop receiving mail is worse than one that
/// changes nothing.
///
/// Imported subscribers stay sendable, marketing included, and there is no
/// re-confirmation pass. The existing signups are treated as valid consent.
/// That is a deliberate acceptance of the fact that nothing recorded what any
/// given subscriber was told they were signing up for, taken against the cost
/// of losing most of the existing list.
///
/// `Imported` therefore survives as provenance only: it says how the row got
/// here, not whether the row may be mailed. Nothing gates on it.
const SENDABLE_STATES: &[&str] = &["confirmed", "imported"];

/// One deliverable recipient.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Recipient {
    /// The subscription this delivery is against. Carried so the caller can
    /// mint a per-recipient unsubscribe link and, later, record the send.
    pub subscription_id: ListSubscriptionId,
    /// `None` for a bare address with no account behind it.
    pub user_id: Option<UserId>,
    pub email: String,
    pub display_name: Option<String>,
}

/// A list and everyone who may currently be mailed on it.
#[derive(Debug, Clone)]
pub struct Audience {
    pub list_id: ListId,
    /// Transactional list nobody may leave, so no unsubscribe footer is owed.
    /// The caller reads this rather than deciding per send, which is what stops
    /// a marketing send from quietly omitting the footer.
    pub required: bool,
    pub recipients: Vec<Recipient>,
}

/// Everyone who may be mailed on a list, and nobody who may not.
///
/// This is the one place the delivery rules live. They were previously spread
/// across each send's own query, which is why suppression was applied
/// consistently (it sat in `send_email_inner`) and nothing else was.
///
/// Applied here, in order:
///   - the subscription state must be sendable (see [`SENDABLE_STATES`]);
///   - the address must not be suppressed, which covers bounces and complaints;
///   - an account subscriber must have a verified, unsuspended account. A bare
///     address has no account to check, and excluding those was a real bug once:
///     an INNER JOIN meant imported subscribers were never mailed.
///
/// Capped at 10,000, matching the query it replaces.
#[tracing::instrument(skip_all)]
pub async fn resolve_audience(pool: &PgPool, list_id: ListId) -> Result<Audience> {
    let required = sqlx::query_scalar::<_, bool>("SELECT required FROM lists WHERE id = $1")
        .bind(list_id)
        .fetch_one(pool)
        .await?;

    let recipients = sqlx::query_as::<_, Recipient>(
        r"
        SELECT ls.id AS subscription_id, u.id AS user_id, u.email, u.display_name
        FROM list_subscriptions ls
        JOIN users u ON u.id = ls.user_id
        WHERE ls.list_id = $1
          AND ls.state = ANY($2)
          AND u.email_verified = true
          AND u.suspended_at IS NULL
          AND LOWER(u.email) NOT IN (SELECT LOWER(email) FROM email_suppressions)
        UNION ALL
        SELECT ls.id AS subscription_id, NULL::uuid AS user_id, ls.email, NULL AS display_name
        FROM list_subscriptions ls
        WHERE ls.list_id = $1
          AND ls.state = ANY($2)
          AND ls.user_id IS NULL
          AND ls.email IS NOT NULL
          AND LOWER(ls.email) NOT IN (SELECT LOWER(email) FROM email_suppressions)
        LIMIT 10000
        ",
    )
    .bind(list_id)
    .bind(SENDABLE_STATES)
    .fetch_all(pool)
    .await?;

    Ok(Audience {
        list_id,
        required,
        recipients,
    })
}

/// The unified list mirroring a legacy per-project list.
///
/// Resolves through `mailing_lists` rather than storing a foreign key, so the
/// old table needs no schema change during the migration and dropping it later
/// leaves nothing dangling.
#[tracing::instrument(skip_all)]
pub async fn list_for_legacy(pool: &PgPool, mailing_list_id: uuid::Uuid) -> Result<Option<ListId>> {
    let id = sqlx::query_scalar::<_, ListId>(
        "SELECT l.id FROM mailing_lists ml \
         JOIN lists l ON l.scope = 'project' AND l.scope_id = ml.project_id AND l.kind = ml.list_type \
         WHERE ml.id = $1",
    )
    .bind(mailing_list_id)
    .fetch_optional(pool)
    .await?;
    Ok(id)
}

/// Count subscriptions on a list in a given state. Exists for the backfill
/// tests and the admin view; the send path uses [`resolve_audience`].
#[tracing::instrument(skip_all)]
pub async fn count_in_state(
    pool: &PgPool,
    list_id: ListId,
    state: SubscriptionState,
) -> Result<i64> {
    let count = sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM list_subscriptions WHERE list_id = $1 AND state = $2",
    )
    .bind(list_id)
    .bind(state.to_string())
    .fetch_one(pool)
    .await?;
    Ok(count)
}

/// One row on the unsubscribe page.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct SubscriptionRow {
    pub subscription_id: ListSubscriptionId,
    pub list_id: ListId,
    pub title: String,
    /// Transactional list. Shown so the page is an honest inventory of what we
    /// send, but it carries no toggle: there is no opting out of a receipt.
    pub required: bool,
    pub state: String,
    /// The creator who owns the list, if one does.
    ///
    /// `None` for a platform list (receipts, security, account mail), which
    /// nobody but us decides. Present for a project list, where the creator is
    /// the other joint controller of the address and the preferences page has
    /// to be able to say so. See `site-docs/public/legal/mailing-list-data-processing.md`.
    pub owner_name: Option<String>,
}

impl SubscriptionRow {
    /// Whether this row is currently receiving mail.
    pub fn subscribed(&self) -> bool {
        SENDABLE_STATES.contains(&self.state.as_str())
    }
}

/// Every list the subscriber behind `subscription_id` is on.
///
/// The page is reached from a link in one email, but it shows all of them:
/// somebody who wants out is rarely asking about the single list that happened
/// to prompt them, and making them hunt for the rest is how "unsubscribe" turns
/// into "mark as spam".
#[tracing::instrument(skip_all)]
pub async fn subscriptions_for_peer(
    pool: &PgPool,
    subscription_id: ListSubscriptionId,
) -> Result<Vec<SubscriptionRow>> {
    let rows = sqlx::query_as::<_, SubscriptionRow>(
        r"
        WITH peer AS (
            SELECT user_id, email FROM list_subscriptions WHERE id = $1
        )
        SELECT ls.id AS subscription_id, l.id AS list_id, l.title, l.required, ls.state,
               COALESCE(NULLIF(u.display_name, ''), u.username) AS owner_name
        FROM list_subscriptions ls
        JOIN lists l ON l.id = ls.list_id
        LEFT JOIN users u ON u.id = l.owner_id
        CROSS JOIN peer
        WHERE (peer.user_id IS NOT NULL AND ls.user_id = peer.user_id)
           OR (peer.email IS NOT NULL AND LOWER(ls.email) = LOWER(peer.email))
        ORDER BY l.required DESC, l.title
        ",
    )
    .bind(subscription_id)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Unsubscribe the peer from every list they may leave.
///
/// Required lists are skipped rather than refused: "unsubscribe from
/// everything" means everything on offer, and a receipt was never on offer.
/// Returns how many moved.
#[tracing::instrument(skip_all)]
pub async fn unsubscribe_peer_from_all(
    pool: &PgPool,
    subscription_id: ListSubscriptionId,
) -> Result<usize> {
    let rows = subscriptions_for_peer(pool, subscription_id).await?;
    let mut moved = 0;
    for row in rows {
        if row.required {
            continue;
        }
        if unsubscribe(pool, row.subscription_id, ConsentEvent::OptOut).await? {
            moved += 1;
        }
    }
    Ok(moved)
}

/// Re-subscribe a row the page had toggled off. Appends an `opt_in`; the
/// `opt_out` before it stays, because the history is the point.
#[tracing::instrument(skip_all)]
pub async fn resubscribe(pool: &PgPool, subscription_id: ListSubscriptionId) -> Result<bool> {
    let mut tx = pool.begin().await?;
    let moved = sqlx::query(
        "UPDATE list_subscriptions \
         SET state = 'confirmed', confirmed_at = NOW(), unsubscribed_at = NULL \
         WHERE id = $1 AND state = 'unsubscribed'",
    )
    .bind(subscription_id)
    .execute(&mut *tx)
    .await?
    .rows_affected()
        > 0;

    if moved {
        sqlx::query(
            "INSERT INTO consent_events (subscription_id, event, evidence) VALUES ($1, 'opt_in', $2)",
        )
        .bind(subscription_id)
        .bind("Re-subscribed from the email preferences page.")
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(moved)
}

/// Whether a subscription's list may be left at all.
#[tracing::instrument(skip_all)]
pub async fn subscription_is_required(
    pool: &PgPool,
    subscription_id: ListSubscriptionId,
) -> Result<bool> {
    let required = sqlx::query_scalar::<_, bool>(
        "SELECT l.required FROM list_subscriptions ls \
         JOIN lists l ON l.id = ls.list_id WHERE ls.id = $1",
    )
    .bind(subscription_id)
    .fetch_optional(pool)
    .await?;
    Ok(required.unwrap_or(false))
}

// ── Per-repo notification lists ──

/// Whether this user has opted out of a repo's notifications.
///
/// The gate is "not opted out" rather than "opted in", which is what keeps the
/// behaviour identical to the account-wide bool it replaces: eligibility is
/// still repo ownership or issue participation, and an absent row still means
/// nothing has been said. Opting in to a repo you have nothing to do with would
/// not get you mail, because it would not make you a participant.
#[tracing::instrument(skip_all)]
pub async fn repo_notifications_muted(
    pool: &PgPool,
    repo_id: uuid::Uuid,
    user_id: UserId,
    kind: ListKind,
) -> Result<bool> {
    let muted = sqlx::query_scalar::<_, bool>(
        "SELECT EXISTS( \
           SELECT 1 FROM list_subscriptions ls \
           JOIN lists l ON l.id = ls.list_id \
           WHERE l.scope = 'repo' AND l.scope_id = $1 AND l.kind = $2 \
             AND ls.user_id = $3 AND ls.state = 'unsubscribed')",
    )
    .bind(repo_id)
    .bind(kind.to_string())
    .bind(user_id)
    .fetch_one(pool)
    .await?;
    Ok(muted)
}

/// Mute or unmute a repo's notifications for one user.
///
/// Muting records an explicit `unsubscribed` row; unmuting moves it back.
/// Either way a consent event is appended, so "when did I turn this off" has an
/// answer.
#[tracing::instrument(skip_all)]
pub async fn set_repo_muted(
    pool: &PgPool,
    repo_id: uuid::Uuid,
    user_id: UserId,
    kind: ListKind,
    muted: bool,
) -> Result<()> {
    let Some(list_id) = find_list(pool, ListScope::Repo, Some(repo_id), kind).await? else {
        return Ok(());
    };

    if muted {
        subscribe(
            pool,
            list_id,
            &Subscriber::User(user_id),
            SubscriptionState::Unsubscribed,
            SubscriptionSource::ProjectPage,
            ConsentEvent::OptOut,
            Some("Muted from the repository page."),
        )
        .await?;
    } else {
        subscribe(
            pool,
            list_id,
            &Subscriber::User(user_id),
            SubscriptionState::Confirmed,
            SubscriptionSource::ProjectPage,
            ConsentEvent::OptIn,
            Some("Unmuted from the repository page."),
        )
        .await?;
    }
    Ok(())
}

// ── Account notification preferences ──
//
// Subscriptions are the whole record. Seven bool columns on `users` held these
// until migration 189; every read moved in 5b and the columns went in 5c, so
// there is one place a preference lives and one place it is read from.
//
// What is left of the columns is their names, in NOTIFICATION_LISTS below:
// unsubscribe links already sent carry them in signed URLs.

/// Platform notification lists, paired with the legacy preference name.
///
/// The second element was the `users.notify_*` column until migration 189
/// dropped them. It survives because unsubscribe links already sitting in
/// inboxes carry those names in their signed URLs, and renaming them would
/// invalidate every link ever sent. `disable_notification` maps them back.
pub const NOTIFICATION_LISTS: &[(&str, &str)] = &[
    ("sale", "notify_sale"),
    ("follower", "notify_follower"),
    ("releases", "notify_release"),
    ("issues", "notify_issues"),
    ("status", "notify_status"),
    ("tip", "notify_tip"),
    ("login", "login_notification_enabled"),
];

/// The `users` column a platform list mirrors, if it mirrors one.
pub fn notification_column_for_kind(kind: &str) -> Option<&'static str> {
    NOTIFICATION_LISTS
        .iter()
        .find(|(k, _)| *k == kind)
        .map(|(_, col)| *col)
}

/// What a preference is when nobody has said otherwise.
///
/// Matches the `users` column defaults, which is where these lived until the
/// reads moved. Only reached if a subscription row is missing, which the 186
/// backfill and the 187 trigger between them should make impossible; the
/// fallback exists so a missing row degrades to the documented default rather
/// than to silence. `notification_rows_exist_for_every_account` is the test
/// that keeps it unreachable.
fn default_enabled(kind: &str) -> bool {
    // Status alerts are the one opt-in: they are platform operations noise, and
    // a new account has not asked for them. Everything else, `invite` included
    // (migration 195, which has no column behind it), is on until turned off.
    kind != "status"
}

/// May this user be sent this kind of notification?
///
/// The single read for every account notification. Each of these used to be a
/// `users.notify_*` column consulted at the send site, which is why the rules
/// could differ per site and why opting out was all-or-nothing.
///
/// A missing subscription falls back to [`default_enabled`] rather than
/// refusing: the failure mode of a bug here should be mail somebody expected,
/// not silence they cannot diagnose.
#[tracing::instrument(skip_all)]
pub async fn may_notify(pool: &PgPool, user_id: UserId, kind: ListKind) -> Result<bool> {
    let kind = kind.to_string();
    let state = sqlx::query_scalar::<_, String>(
        "SELECT ls.state FROM list_subscriptions ls \
         JOIN lists l ON l.id = ls.list_id \
         WHERE l.scope = 'platform' AND l.kind = $1 AND ls.user_id = $2",
    )
    .bind(&kind)
    .bind(user_id)
    .fetch_optional(pool)
    .await?;

    Ok(match state {
        Some(s) => SENDABLE_STATES.contains(&s.as_str()),
        None => {
            tracing::warn!(
                user_id = %user_id, kind = %kind,
                "no notification subscription row; falling back to the default"
            );
            default_enabled(&kind)
        }
    })
}

/// Every notification preference for one user, for the settings screen.
///
/// One query rather than seven `may_notify` calls, because the account tab
/// renders all of them at once. Missing rows fall back to the same defaults
/// `may_notify` uses, so the two cannot disagree about an account the trigger
/// somehow missed.
#[derive(Debug, Clone)]
pub struct NotificationPrefs {
    pub sale: bool,
    pub follower: bool,
    pub release: bool,
    pub login: bool,
    pub issues: bool,
    pub status: bool,
    pub tip: bool,
    pub invite: bool,
}

#[tracing::instrument(skip_all)]
pub async fn notification_prefs(pool: &PgPool, user_id: UserId) -> Result<NotificationPrefs> {
    let rows = sqlx::query_as::<_, (String, String)>(
        "SELECT l.kind, ls.state FROM list_subscriptions ls \
         JOIN lists l ON l.id = ls.list_id \
         WHERE l.scope = 'platform' AND ls.user_id = $1",
    )
    .bind(user_id)
    .fetch_all(pool)
    .await?;

    let enabled = |kind: &str| {
        rows.iter().find(|(k, _)| k == kind).map_or_else(
            || default_enabled(kind),
            |(_, state)| SENDABLE_STATES.contains(&state.as_str()),
        )
    };

    Ok(NotificationPrefs {
        sale: enabled("sale"),
        follower: enabled("follower"),
        release: enabled("releases"),
        login: enabled("login"),
        issues: enabled("issues"),
        status: enabled("status"),
        tip: enabled("tip"),
        invite: enabled("invite"),
    })
}

/// Point a user's notification subscription at `enabled`, appending the consent
/// event that goes with it. Called after the column is written.
#[tracing::instrument(skip_all)]
pub async fn sync_notification_subscription(
    pool: &PgPool,
    user_id: UserId,
    kind: &str,
    enabled: bool,
) -> Result<()> {
    let Some(list_id) = sqlx::query_scalar::<_, ListId>(
        "SELECT id FROM lists WHERE scope = 'platform' AND kind = $1",
    )
    .bind(kind)
    .fetch_optional(pool)
    .await?
    else {
        return Ok(());
    };

    if enabled {
        subscribe(
            pool,
            list_id,
            &Subscriber::User(user_id),
            SubscriptionState::Confirmed,
            SubscriptionSource::Admin,
            ConsentEvent::OptIn,
            Some("Enabled from account notification settings."),
        )
        .await?;
        return Ok(());
    }

    let existing = sqlx::query_scalar::<_, ListSubscriptionId>(
        "SELECT id FROM list_subscriptions WHERE list_id = $1 AND user_id = $2",
    )
    .bind(list_id)
    .bind(user_id)
    .fetch_optional(pool)
    .await?;

    match existing {
        Some(subscription_id) => {
            unsubscribe(pool, subscription_id, ConsentEvent::OptOut).await?;
        }
        None => {
            // No row yet (an account created since the backfill). Record the
            // "no" rather than leaving it absent, so it reads as a choice
            // rather than as never having been asked.
            subscribe(
                pool,
                list_id,
                &Subscriber::User(user_id),
                SubscriptionState::Unsubscribed,
                SubscriptionSource::Admin,
                ConsentEvent::OptOut,
                Some("Disabled from account notification settings."),
            )
            .await?;
        }
    }
    Ok(())
}

/// The `users` column behind a subscription, if it has one.
///
/// Used by the preferences page: a toggle there has to reach the column, or the
/// send that reads the column will ignore it.
#[tracing::instrument(skip_all)]
pub async fn notification_column_for_subscription(
    pool: &PgPool,
    subscription_id: ListSubscriptionId,
) -> Result<Option<(UserId, &'static str)>> {
    let row = sqlx::query_as::<_, (Option<UserId>, String, String)>(
        "SELECT ls.user_id, l.scope, l.kind FROM list_subscriptions ls \
         JOIN lists l ON l.id = ls.list_id WHERE ls.id = $1",
    )
    .bind(subscription_id)
    .fetch_optional(pool)
    .await?;

    let Some((Some(user_id), scope, kind)) = row else {
        return Ok(None);
    };
    if scope != "platform" {
        return Ok(None);
    }
    Ok(notification_column_for_kind(&kind).map(|col| (user_id, col)))
}

// ── Mirroring the legacy tables ──
//
// `mailing_lists` / `mailing_list_subscribers` are still what the product
// writes to, and `resolve_audience` is what sends now read. Every legacy write
// therefore has to reach here, or a subscriber added after the migration is one
// no send can see.
//
// These propagate their errors rather than logging and continuing. A subscribe
// that does not reach the send path is a broken subscribe, and the failure
// should be visible where it happened rather than at the next announcement.

/// Mirror a legacy project list into `lists`.
#[tracing::instrument(skip_all)]
pub async fn mirror_legacy_list(
    pool: &PgPool,
    project_id: uuid::Uuid,
    kind: ListKind,
    title: &str,
) -> Result<()> {
    sqlx::query(
        "INSERT INTO lists (scope, scope_id, kind, title, required, owner_id) \
         SELECT 'project', $1, $2, $3, FALSE, p.user_id FROM projects p WHERE p.id = $1 \
         ON CONFLICT DO NOTHING",
    )
    .bind(project_id)
    .bind(kind.to_string())
    .bind(title)
    .execute(pool)
    .await?;
    Ok(())
}

/// Mirror a legacy subscribe.
///
/// A subscribe through the product is a real act, so it lands `confirmed` with
/// an `opt_in` event, unlike the backfill's `imported`. `evidence` records what
/// the person was doing at the time, which is the difference between consent we
/// can show and consent we assert.
#[tracing::instrument(skip_all)]
pub async fn mirror_legacy_subscribe(
    pool: &PgPool,
    mailing_list_id: uuid::Uuid,
    subscriber: &Subscriber,
    state: SubscriptionState,
    source: SubscriptionSource,
    evidence: Option<&str>,
) -> Result<()> {
    let Some(list_id) = list_for_legacy(pool, mailing_list_id).await? else {
        // The legacy list predates its mirror. Create-list mirroring runs first
        // for anything made since the migration, so this means a list that was
        // never backfilled, which is a bug worth seeing rather than skipping.
        return Err(crate::error::AppError::Internal(anyhow::anyhow!(
            "legacy mailing list {mailing_list_id} has no unified list"
        )));
    };
    let event = match state {
        SubscriptionState::Imported => ConsentEvent::Import,
        _ => ConsentEvent::OptIn,
    };
    subscribe(pool, list_id, subscriber, state, source, event, evidence).await?;
    Ok(())
}

/// Mirror a legacy unsubscribe for an account subscriber.
///
/// The most important mirror of the three: a missed unsubscribe means mailing
/// somebody who asked us not to.
#[tracing::instrument(skip_all)]
pub async fn mirror_legacy_unsubscribe_user(
    pool: &PgPool,
    mailing_list_id: uuid::Uuid,
    user_id: UserId,
) -> Result<()> {
    let Some(list_id) = list_for_legacy(pool, mailing_list_id).await? else {
        return Ok(());
    };
    mark_unsubscribed(pool, list_id, &Subscriber::User(user_id)).await
}

/// Mirror a legacy unsubscribe for a bare address.
#[tracing::instrument(skip_all)]
pub async fn mirror_legacy_unsubscribe_email(
    pool: &PgPool,
    mailing_list_id: uuid::Uuid,
    email: &str,
) -> Result<()> {
    let Some(list_id) = list_for_legacy(pool, mailing_list_id).await? else {
        return Ok(());
    };
    mark_unsubscribed(pool, list_id, &Subscriber::Email(email.to_string())).await
}

/// Mirror an unsubscribe from every list on a project (the unfollow path).
#[tracing::instrument(skip_all)]
pub async fn mirror_legacy_unsubscribe_project(
    pool: &PgPool,
    project_id: uuid::Uuid,
    user_id: UserId,
) -> Result<()> {
    let ids = sqlx::query_scalar::<_, ListId>(
        "SELECT id FROM lists WHERE scope = 'project' AND scope_id = $1",
    )
    .bind(project_id)
    .fetch_all(pool)
    .await?;
    for list_id in ids {
        mark_unsubscribed(pool, list_id, &Subscriber::User(user_id)).await?;
    }
    Ok(())
}

/// Move a subscription to `unsubscribed` and append the opt-out, by identity
/// rather than by subscription id. No-op when there is nothing subscribed.
async fn mark_unsubscribed(pool: &PgPool, list_id: ListId, subscriber: &Subscriber) -> Result<()> {
    let existing =
        match subscriber {
            Subscriber::User(id) => {
                sqlx::query_scalar::<_, ListSubscriptionId>(
                    "SELECT id FROM list_subscriptions WHERE list_id = $1 AND user_id = $2",
                )
                .bind(list_id)
                .bind(id)
                .fetch_optional(pool)
                .await?
            }
            Subscriber::Email(addr) => sqlx::query_scalar::<_, ListSubscriptionId>(
                "SELECT id FROM list_subscriptions WHERE list_id = $1 AND LOWER(email) = LOWER($2)",
            )
            .bind(list_id)
            .bind(addr)
            .fetch_optional(pool)
            .await?,
        };
    if let Some(subscription_id) = existing {
        unsubscribe(pool, subscription_id, ConsentEvent::OptOut).await?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn scope_and_kind_round_trip() {
        for s in [
            ListScope::Platform,
            ListScope::Project,
            ListScope::Repo,
            ListScope::Creator,
        ] {
            assert_eq!(s.to_string().parse::<ListScope>().unwrap(), s);
        }
        for k in [ListKind::Content, ListKind::Marketing, ListKind::Issues] {
            assert_eq!(k.to_string().parse::<ListKind>().unwrap(), k);
        }
    }

    /// Every `ListKind` is accepted by the database.
    ///
    /// A kind lives in two places: this enum and the `lists_kind_check`
    /// constraint. Adding it to only the enum compiles, passes every unit test,
    /// and then fails at INSERT against a deployed database, which is a long
    /// way from the edit that caused it. So the constraint is read back here.
    ///
    /// Reads the last migration that redefines the constraint, since each one
    /// replaces the previous in full (186, then 195).
    #[test]
    fn every_kind_is_allowed_by_the_check_constraint() {
        let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
        let mut files: Vec<_> = std::fs::read_dir(&dir)
            .expect("migrations directory")
            .filter_map(|e| e.ok().map(|e| e.path()))
            .filter(|p| {
                std::fs::read_to_string(p).is_ok_and(|s| s.contains("lists_kind_check CHECK"))
            })
            .collect();
        files.sort();
        let newest = files
            .last()
            .expect("some migration defines lists_kind_check");
        let sql = std::fs::read_to_string(newest).expect("readable migration");

        let clause = sql
            .split_once("lists_kind_check CHECK (kind IN (")
            .expect("the constraint has the expected shape")
            .1
            .split_once("))")
            .expect("the constraint list is closed")
            .0;
        let allowed: Vec<&str> = clause
            .split(',')
            .map(|s| s.trim().trim_matches('\'').trim())
            .filter(|s| !s.is_empty())
            .collect();

        for kind in ListKind::ALL {
            let s = kind.to_string();
            assert!(
                allowed.contains(&s.as_str()),
                "ListKind::{kind:?} (\"{s}\") is not in the lists_kind_check constraint. \
                 Adding a kind takes a migration as well as an enum variant, or the first \
                 insert of one fails on a deployed database.",
            );
        }
        assert_eq!(
            allowed.len(),
            ListKind::ALL.len(),
            "the constraint allows {allowed:?}, which is not the set ListKind names. A kind \
             the database accepts but the enum cannot represent is unreachable from the code.",
        );
    }

    /// Which states receive mail is a policy, and a settled one. Changing this
    /// set changes who gets email, so it should be an edit somebody made on
    /// purpose rather than a line that moved during a refactor.
    #[test]
    fn sendable_states_are_the_agreed_set() {
        assert_eq!(
            SENDABLE_STATES,
            &["confirmed", "imported"],
            "'imported' is sendable by decision (GoingsOn 04a882b4, 2026-08-06): no \
             re-confirmation pass, the existing signups count as consent. Dropping it \
             silently stops mail for most of the list, so it needs a new decision and \
             not just a diff"
        );
    }

    /// The strings are a database CHECK constraint, so a rename here that is
    /// not matched by a migration fails at insert rather than at compile time.
    #[test]
    fn state_and_event_strings_match_the_check_constraints() {
        assert_eq!(SubscriptionState::Imported.to_string(), "imported");
        assert_eq!(SubscriptionState::Unsubscribed.to_string(), "unsubscribed");
        assert_eq!(SubscriptionSource::LandingForm.to_string(), "landing_form");
        assert_eq!(ConsentEvent::AdminRemoval.to_string(), "admin_removal");
        assert_eq!(ConsentEvent::Import.to_string(), "import");
    }
}