headless_lms_models/

credit_registration_account_linking_emails.rs

//! Dedup ledger for account-linking mails.
//!
//! Keyed on the Sisu person id plus the recipient address: at send time there is no account of ours
//! to key on, and the student number changes when a student moves between programmes. A row is
//! written when the right to mail is claimed, before a delivery exists, so a crash between the two
//! phases cannot mail twice. A row holds its address only while its link can still be used; once
//! the link has expired or been used, a later mail to the address replaces it.
use std::collections::{HashMap, HashSet};

use secrecy::ExposeSecret;
use utoipa::ToSchema;

use crate::email_deliveries::{
    EmailSendStatus, EmailSendStatusFacts, EmailSendStatusReport, derive_email_send_status,
    get_send_statuses, is_hard_send_failure,
};
use crate::prelude::*;

#[derive(Debug, Clone)]
pub struct CreditRegistrationAccountLinkingEmail {
    pub id: Uuid,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub deleted_at: Option<DateTime<Utc>>,
    pub student_number: DbSecret,
    pub sisu_person_id: DbSecret,
    pub course_id: Uuid,
    pub emailed_to: DbSecret,
    pub student_number_verification_token_id: Option<Uuid>,
    pub email_delivery_id: Option<Uuid>,
    pub sent_at: DateTime<Utc>,
    /// Set once this row gave up its place in the dedup key: to a later mail to the same address,
    /// or when the person was unlinked.
    pub replaced_at: Option<DateTime<Utc>>,
}

#[derive(Debug, Clone)]
pub struct NewAccountLinkingEmail {
    pub student_number: DbSecret,
    pub sisu_person_id: DbSecret,
    pub course_id: Uuid,
    pub emailed_to: DbSecret,
    pub student_number_verification_token_id: Option<Uuid>,
    pub email_delivery_id: Option<Uuid>,
}

/// Claims the right to mail this (person, course, address); `None` means another row holds the
/// address and the caller must not send. Call in the transaction that mints the token, so a refused
/// claim leaves no usable link.
pub async fn claim_send_slot(
    conn: &mut PgConnection,
    new: &NewAccountLinkingEmail,
) -> ModelResult<Option<Uuid>> {
    let res = sqlx::query!(
        r#"
INSERT INTO credit_registration_account_linking_emails (
    student_number,
    sisu_person_id,
    course_id,
    emailed_to,
    student_number_verification_token_id,
    email_delivery_id
  )
VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT DO NOTHING
RETURNING id
        "#,
        new.student_number.expose_secret(),
        new.sisu_person_id.expose_secret(),
        new.course_id,
        new.emailed_to.expose_secret(),
        new.student_number_verification_token_id,
        new.email_delivery_id,
    )
    .fetch_optional(conn)
    .await?;
    Ok(res.map(|r| r.id))
}

/// Moves a mail's `sent_at` `age` into the past, for seeding mails sent before today.
///
/// Shifts the row rather than the clock because the caps compare against Postgres `now()`.
pub async fn backdate(conn: &mut PgConnection, id: Uuid, age: chrono::Duration) -> ModelResult<()> {
    sqlx::query!(
        "UPDATE credit_registration_account_linking_emails SET sent_at = now() - $2::interval WHERE id = $1",
        id,
        age as chrono::Duration
    )
    .execute(conn)
    .await?;
    Ok(())
}

/// As much of one existing mail as the dedup guard and the rate caps need to decide on a person.
#[derive(Debug, Clone)]
pub struct ExistingLinkingMailFact {
    pub sisu_person_id: DbSecret,
    pub course_id: Uuid,
    pub emailed_to: DbSecret,
    pub sent_at: DateTime<Utc>,
    /// Whether the row still holds its address: not replaced, and its link can still be used. A row
    /// with no token holds it for good.
    pub holds_address: bool,
}

/// Every live mail these people have ever been sent, any course: one query stands in for the
/// quiet-period count, the per-course count and the dedup check of a whole batch of candidates.
pub async fn get_existing_facts_for_persons(
    conn: &mut PgConnection,
    sisu_person_ids: &[String],
) -> ModelResult<Vec<ExistingLinkingMailFact>> {
    let res = sqlx::query_as!(
        ExistingLinkingMailFact,
        r#"
SELECT e.sisu_person_id AS "sisu_person_id!",
  e.course_id AS "course_id!",
  e.emailed_to AS "emailed_to!",
  e.sent_at AS "sent_at!",
  (
    e.replaced_at IS NULL
    AND (
      e.student_number_verification_token_id IS NULL
      OR is_usable_verification_token(t)
    )
  ) AS "holds_address!"
FROM credit_registration_account_linking_emails e
  LEFT JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
WHERE e.sisu_person_id = ANY($1::text [])
  AND e.deleted_at IS NULL
        "#,
        sisu_person_ids
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

/// How many linking mails one course has sent each of these people, replaced ones included.
pub async fn count_mails_per_person_for_course(
    conn: &mut PgConnection,
    sisu_person_ids: &[String],
    course_id: Uuid,
) -> ModelResult<HashMap<String, i64>> {
    let res = sqlx::query!(
        r#"
SELECT sisu_person_id AS "sisu_person_id!",
  COUNT(*) AS "mail_count!"
FROM credit_registration_account_linking_emails
WHERE sisu_person_id = ANY($1::text [])
  AND course_id = $2
  AND deleted_at IS NULL
GROUP BY sisu_person_id
        "#,
        sisu_person_ids,
        course_id
    )
    .fetch_all(conn)
    .await?;
    Ok(res
        .into_iter()
        .map(|row| {
            (
                row.sisu_person_id.expose_secret().to_owned(),
                row.mail_count,
            )
        })
        .collect())
}

/// Marks the rows these new slots would collide with in the dedup key as replaced, where their link
/// can no longer be used, so the slots can be claimed. Rows holding their address are left alone.
pub async fn replace_lapsed_slots(
    conn: &mut PgConnection,
    new: &[NewAccountLinkingEmail],
) -> ModelResult<()> {
    if new.is_empty() {
        return Ok(());
    }
    let sisu_person_ids: Vec<String> = new
        .iter()
        .map(|n| n.sisu_person_id.expose_secret().to_owned())
        .collect();
    let course_ids: Vec<Uuid> = new.iter().map(|n| n.course_id).collect();
    let emailed_tos: Vec<String> = new
        .iter()
        .map(|n| n.emailed_to.expose_secret().to_owned())
        .collect();
    sqlx::query!(
        r#"
UPDATE credit_registration_account_linking_emails e
SET replaced_at = now()
FROM UNNEST($1::text [], $2::uuid [], $3::text []) AS slot(sisu_person_id, course_id, emailed_to)
WHERE e.sisu_person_id = slot.sisu_person_id
  AND e.course_id = slot.course_id
  AND LOWER(e.emailed_to) = LOWER(slot.emailed_to)
  AND e.deleted_at IS NULL
  AND e.replaced_at IS NULL
  AND e.student_number_verification_token_id IS NOT NULL
  AND NOT EXISTS (
    SELECT 1
    FROM student_number_verification_tokens t
    WHERE t.id = e.student_number_verification_token_id
      AND is_usable_verification_token(t)
  )
        "#,
        &sisu_person_ids,
        &course_ids,
        &emailed_tos,
    )
    .execute(conn)
    .await?;
    Ok(())
}

/// Batched form of [`claim_send_slot`], keyed by token: returns the
/// `student_number_verification_token_id` of every row actually inserted, so the caller can tell
/// which candidates lost the race to the unique index.
///
/// `token_ids[i]` is the token already minted for `new[i]`, taken separately rather than read from
/// `new[i].student_number_verification_token_id`: that field stays `Option` because it really can be
/// absent on the single-row [`claim_send_slot`], but a batched claim always has one.
pub async fn claim_send_slots(
    conn: &mut PgConnection,
    new: &[NewAccountLinkingEmail],
    token_ids: &[Uuid],
) -> ModelResult<HashSet<Uuid>> {
    if new.is_empty() {
        return Ok(HashSet::new());
    }
    let student_numbers: Vec<String> = new
        .iter()
        .map(|n| n.student_number.expose_secret().to_owned())
        .collect();
    let sisu_person_ids: Vec<String> = new
        .iter()
        .map(|n| n.sisu_person_id.expose_secret().to_owned())
        .collect();
    let course_ids: Vec<Uuid> = new.iter().map(|n| n.course_id).collect();
    let emailed_tos: Vec<String> = new
        .iter()
        .map(|n| n.emailed_to.expose_secret().to_owned())
        .collect();

    let claimed = sqlx::query_scalar!(
        r#"
INSERT INTO credit_registration_account_linking_emails (
    student_number,
    sisu_person_id,
    course_id,
    emailed_to,
    student_number_verification_token_id
  )
SELECT * FROM UNNEST($1::text [], $2::text [], $3::uuid [], $4::text [], $5::uuid []) ON CONFLICT DO NOTHING
RETURNING student_number_verification_token_id AS "token_id!"
        "#,
        &student_numbers,
        &sisu_person_ids,
        &course_ids,
        &emailed_tos,
        token_ids,
    )
    .fetch_all(conn)
    .await?;
    Ok(claimed.into_iter().collect())
}

/// This course's newest linking mail for each of these Sisu people, keyed by person id.
pub async fn get_latest_by_course_and_persons(
    conn: &mut PgConnection,
    course_id: Uuid,
    sisu_person_ids: &[String],
) -> ModelResult<HashMap<String, CreditRegistrationAccountLinkingEmail>> {
    let res = sqlx::query_as!(
        CreditRegistrationAccountLinkingEmail,
        r#"
SELECT DISTINCT ON (sisu_person_id) *
FROM credit_registration_account_linking_emails
WHERE course_id = $1
  AND sisu_person_id = ANY($2::text [])
  AND deleted_at IS NULL
ORDER BY sisu_person_id,
  sent_at DESC
        "#,
        course_id,
        sisu_person_ids,
    )
    .fetch_all(conn)
    .await?;
    Ok(res
        .into_iter()
        .map(|row| (row.sisu_person_id.expose_secret().to_owned(), row))
        .collect())
}

/// This course's linking mails for one student number, newest first.
///
/// Keyed on the number the study registry gave us, not on a verified link: the recipients of a
/// linking mail are exactly the population that has none.
pub async fn get_by_course_id_and_student_number(
    conn: &mut PgConnection,
    course_id: Uuid,
    student_number: &str,
) -> ModelResult<Vec<CreditRegistrationAccountLinkingEmail>> {
    let res = sqlx::query_as!(
        CreditRegistrationAccountLinkingEmail,
        r#"
SELECT *
FROM credit_registration_account_linking_emails
WHERE course_id = $1
  AND student_number = $2
  AND deleted_at IS NULL
ORDER BY sent_at DESC
        "#,
        course_id,
        student_number,
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

pub async fn get_by_sisu_person_id(
    conn: &mut PgConnection,
    sisu_person_id: &str,
) -> ModelResult<Vec<CreditRegistrationAccountLinkingEmail>> {
    let res = sqlx::query_as!(
        CreditRegistrationAccountLinkingEmail,
        r#"
SELECT *
FROM credit_registration_account_linking_emails
WHERE sisu_person_id = $1
  AND deleted_at IS NULL
ORDER BY sent_at DESC
        "#,
        sisu_person_id
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

/// How many mails this person has had for this course, replaced ones included. Read against the
/// lifetime cap when an admin asks for a resend.
pub async fn count_sent_for_person_and_course(
    conn: &mut PgConnection,
    sisu_person_id: &str,
    course_id: Uuid,
) -> ModelResult<i64> {
    let count = sqlx::query_scalar!(
        r#"
SELECT COUNT(*) AS "count!"
FROM credit_registration_account_linking_emails
WHERE sisu_person_id = $1
  AND course_id = $2
  AND deleted_at IS NULL
        "#,
        sisu_person_id,
        course_id,
    )
    .fetch_one(conn)
    .await?;
    Ok(count)
}

/// A claimed slot with no delivery yet, and everything the mail needs to be written.
#[derive(Debug, Clone)]
pub struct LinkingMailToQueue {
    pub id: Uuid,
    pub emailed_to: DbSecret,
    pub student_number: DbSecret,
    pub first_names: Option<DbSecret>,
    /// Mailed as part of the link, so the recipient can prove the address is theirs.
    pub token: DbSecret,
    pub course_name: String,
    pub course_language_code: String,
}

/// Claims slots whose mail has not been queued yet, oldest first.
///
/// Locks them, so the caller must hold a transaction: the delivery insert is not idempotent, and two
/// iterations claiming one slot would queue the same mail twice.
///
/// A retired, used or expired token is skipped rather than mailed — a dead link would spend one of
/// the recipient's mails for this course on nothing — but its slot stays, counting against the caps.
pub async fn claim_unqueued(
    conn: &mut PgConnection,
    limit: i64,
    course_id: Option<Uuid>,
) -> ModelResult<Vec<LinkingMailToQueue>> {
    let res = sqlx::query_as!(
        LinkingMailToQueue,
        r#"
SELECT e.id AS "id!",
  e.emailed_to AS "emailed_to!",
  e.student_number AS "student_number!",
  t.first_names AS "first_names?",
  t.token AS "token!: DbSecret",
  c.name AS "course_name!",
  c.language_code AS "course_language_code!"
FROM credit_registration_account_linking_emails e
  JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
  JOIN courses c ON c.id = e.course_id AND c.deleted_at IS NULL
WHERE e.email_delivery_id IS NULL
  AND e.deleted_at IS NULL
  AND is_usable_verification_token(t)
  AND ($2::uuid IS NULL OR e.course_id = $2)
ORDER BY e.sent_at
FOR UPDATE OF e SKIP LOCKED
LIMIT $1
        "#,
        limit,
        course_id,
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

/// Records which delivery carries this mail, which is also what takes the slot out of the queue.
pub async fn set_email_delivery_id(
    conn: &mut PgConnection,
    id: Uuid,
    email_delivery_id: Uuid,
) -> ModelResult<()> {
    sqlx::query!(
        r#"
UPDATE credit_registration_account_linking_emails
SET email_delivery_id = $2
WHERE id = $1
  AND deleted_at IS NULL
        "#,
        id,
        email_delivery_id,
    )
    .execute(conn)
    .await?;
    Ok(())
}

/// Whether the link a linking mail carries can still be opened.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum LinkingMailLinkState {
    /// Also a mail with no token, which has no link to lapse.
    Usable,
    Used,
    /// Expired, or retired unused.
    Lapsed,
}

impl LinkingMailLinkState {
    /// The state a mail's link is in, given whether it can still be used and whether it was used.
    fn of(is_usable: bool, is_used: bool) -> Self {
        if is_usable {
            Self::Usable
        } else if is_used {
            Self::Used
        } else {
            Self::Lapsed
        }
    }
}

/// The state of each mail's link, by mail id.
pub async fn get_link_states(
    conn: &mut PgConnection,
    mail_ids: &[Uuid],
) -> ModelResult<HashMap<Uuid, LinkingMailLinkState>> {
    let rows = sqlx::query!(
        r#"
SELECT e.id AS "id!",
  (
    e.student_number_verification_token_id IS NULL
    OR is_usable_verification_token(t)
  ) AS "is_usable!",
  t.used_at IS NOT NULL AS "is_used!"
FROM credit_registration_account_linking_emails e
  LEFT JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
WHERE e.id = ANY($1::uuid [])
        "#,
        mail_ids,
    )
    .fetch_all(conn)
    .await?;
    Ok(rows
        .into_iter()
        .map(|row| (row.id, LinkingMailLinkState::of(row.is_usable, row.is_used)))
        .collect())
}

/// What we can honestly say about each linking mail. A slot with no delivery yet is `queued`:
/// anything else would claim a send attempt that never happened.
pub async fn get_send_status_reports(
    conn: &mut PgConnection,
    ids: &[Uuid],
) -> ModelResult<HashMap<Uuid, EmailSendStatusReport>> {
    let rows = sqlx::query!(
        r#"
SELECT id,
  email_delivery_id
FROM credit_registration_account_linking_emails
WHERE id = ANY($1::uuid [])
  AND deleted_at IS NULL
        "#,
        ids
    )
    .fetch_all(&mut *conn)
    .await?;
    let delivery_ids: Vec<Uuid> = rows
        .iter()
        .filter_map(|row| row.email_delivery_id)
        .collect();
    let mut deliveries = get_send_statuses(conn, &delivery_ids).await?;
    let res = rows
        .into_iter()
        .map(|row| {
            let report = row
                .email_delivery_id
                .and_then(|id| deliveries.remove(&id))
                .unwrap_or_else(not_handed_over_yet);
            (row.id, report)
        })
        .collect();
    Ok(res)
}

/// What a claimed slot with no delivery reports. Public so a caller falling back to a default says
/// the same thing [`get_send_status_reports`] would have.
pub fn not_handed_over_yet() -> EmailSendStatusReport {
    EmailSendStatusReport {
        email_send_status: EmailSendStatus::Queued,
        sent_at: None,
        last_attempt_at: None,
        retry_count: 0,
        next_retry_at: None,
        failure_code: None,
        failure_is_transient: None,
    }
}

/// Mails claimed in the window, whatever course they belong to. One row is one address, which is
/// what the send-rate caps govern.
pub async fn count_sent_since(conn: &mut PgConnection, since: DateTime<Utc>) -> ModelResult<i64> {
    let count = sqlx::query_scalar!(
        r#"
SELECT COUNT(*) AS "count!"
FROM credit_registration_account_linking_emails
WHERE sent_at >= $1
  AND deleted_at IS NULL
        "#,
        since,
    )
    .fetch_one(conn)
    .await?;
    Ok(count)
}

/// Every mail in a window, bucketed the way [`crate::email_deliveries::derive_email_send_status`]
/// does. Computed in SQL from the same facts rather than from its output, so the two cannot drift
/// on what counts as failed.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct LinkingMailSendStatusTotals {
    pub mails_in_window: i64,
    /// Claimed, with no delivery yet: the `link-emails` phase has not queued them.
    pub waiting_for_link_emails: i64,
    /// Queued as a delivery the email worker has not attempted yet.
    pub waiting_for_email_worker: i64,
    pub retrying: i64,
    pub sent: i64,
    pub send_failed: i64,
    /// Of `mails_in_window`, those whose link has been used.
    pub used: i64,
    /// `None` when nothing failed within the window.
    pub last_send_failed_at: Option<DateTime<Utc>>,
}

pub async fn get_send_status_totals_since(
    conn: &mut PgConnection,
    since: DateTime<Utc>,
    now: DateTime<Utc>,
) -> ModelResult<LinkingMailSendStatusTotals> {
    struct Row {
        sent_at: DateTime<Utc>,
        email_delivery_id: Option<Uuid>,
        delivery_sent: Option<bool>,
        retryable: Option<bool>,
        first_failed_at: Option<DateTime<Utc>>,
        retry_count: Option<i32>,
        is_used: bool,
    }
    let rows = sqlx::query_as!(
        Row,
        r#"
SELECT
  e.sent_at AS "sent_at!",
  e.email_delivery_id,
  ed.sent AS "delivery_sent?",
  ed.retryable AS "retryable?",
  ed.first_failed_at AS "first_failed_at?",
  ed.retry_count AS "retry_count?",
  COALESCE(t.used_at IS NOT NULL, FALSE) AS "is_used!"
FROM credit_registration_account_linking_emails e
  LEFT JOIN email_deliveries ed ON ed.id = e.email_delivery_id AND ed.deleted_at IS NULL
  LEFT JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
WHERE e.sent_at >= $1
  AND e.deleted_at IS NULL
        "#,
        since,
    )
    .fetch_all(conn)
    .await?;

    let mut totals = LinkingMailSendStatusTotals {
        mails_in_window: rows.len() as i64,
        ..Default::default()
    };
    for row in rows {
        totals.used += i64::from(row.is_used);
        let status = row.email_delivery_id.map(|_| {
            let facts = EmailSendStatusFacts {
                sent: row.delivery_sent.unwrap_or(false),
                retryable: row.retryable.unwrap_or(false),
                retry_count: row.retry_count.unwrap_or(0),
                next_retry_at: None,
                first_failed_at: row.first_failed_at,
                last_attempt_at: None,
                failure_code: None,
                failure_is_transient: None,
            };
            derive_email_send_status(&facts, now).email_send_status
        });
        match status {
            None => totals.waiting_for_link_emails += 1,
            Some(EmailSendStatus::Queued) => totals.waiting_for_email_worker += 1,
            Some(EmailSendStatus::Retrying) => totals.retrying += 1,
            Some(EmailSendStatus::Sent) => totals.sent += 1,
            Some(EmailSendStatus::SendFailed) => {
                totals.send_failed += 1;
                totals.last_send_failed_at = Some(
                    totals
                        .last_send_failed_at
                        .map_or(row.sent_at, |prev| prev.max(row.sent_at)),
                );
            }
        }
    }
    Ok(totals)
}

#[derive(Debug, Clone, PartialEq)]
pub struct LinkingMailFailureDomain {
    pub domain: String,
    pub count: i64,
}

/// Domains behind a hard send failure in the window, worst first. Same predicate as
/// [`get_send_status_totals_since`], enforced by both calling
/// [`crate::email_deliveries::is_hard_send_failure`] rather than each repeating the condition.
pub async fn get_send_failure_domains_since(
    conn: &mut PgConnection,
    since: DateTime<Utc>,
    now: DateTime<Utc>,
) -> ModelResult<Vec<LinkingMailFailureDomain>> {
    struct Row {
        emailed_to: DbSecret,
        retryable: bool,
        first_failed_at: Option<DateTime<Utc>>,
    }
    let rows = sqlx::query_as!(
        Row,
        r#"
SELECT e.emailed_to, ed.retryable, ed.first_failed_at
FROM credit_registration_account_linking_emails e
  JOIN email_deliveries ed ON ed.id = e.email_delivery_id AND ed.deleted_at IS NULL
WHERE e.sent_at >= $1
  AND e.deleted_at IS NULL
  AND position('@' IN e.emailed_to) > 0
  AND NOT ed.sent
        "#,
        since,
    )
    .fetch_all(conn)
    .await?;

    let mut counts: HashMap<String, i64> = HashMap::new();
    for row in rows {
        if !is_hard_send_failure(row.retryable, row.first_failed_at, now) {
            continue;
        }
        let emailed_to = row.emailed_to.expose_secret();
        let Some(at) = emailed_to.find('@') else {
            continue;
        };
        *counts.entry(emailed_to[at + 1..].to_string()).or_insert(0) += 1;
    }

    let mut domains: Vec<LinkingMailFailureDomain> = counts
        .into_iter()
        .map(|(domain, count)| LinkingMailFailureDomain { domain, count })
        .collect();
    domains.sort_by(|a, b| b.count.cmp(&a.count).then_with(|| a.domain.cmp(&b.domain)));
    Ok(domains)
}

/// Hard send failures for one course, all time. Same predicate as [`get_send_status_totals_since`],
/// narrowed to a course instead of a time window.
pub async fn count_send_failed_for_course(
    conn: &mut PgConnection,
    course_id: Uuid,
    now: DateTime<Utc>,
) -> ModelResult<i64> {
    struct Row {
        retryable: bool,
        first_failed_at: Option<DateTime<Utc>>,
    }
    let rows = sqlx::query_as!(
        Row,
        r#"
SELECT ed.retryable, ed.first_failed_at
FROM credit_registration_account_linking_emails e
  JOIN email_deliveries ed ON ed.id = e.email_delivery_id AND ed.deleted_at IS NULL
WHERE e.course_id = $1
  AND e.deleted_at IS NULL
  AND NOT ed.sent
        "#,
        course_id,
    )
    .fetch_all(conn)
    .await?;
    Ok(rows
        .into_iter()
        .filter(|row| is_hard_send_failure(row.retryable, row.first_failed_at, now))
        .count() as i64)
}

/// One person and course mailed to the cap without a single claim: the stale-address population,
/// which is how "the student is ignoring us" is told from "the address Sisu holds is dead".
#[derive(Debug, Clone)]
pub struct StaleUnclaimedLinkingMails {
    pub student_number: DbSecret,
    pub sisu_person_id: DbSecret,
    pub course_id: Uuid,
    pub course_name: String,
    pub mail_count: i64,
    pub first_sent_at: DateTime<Utc>,
    pub last_sent_at: DateTime<Utc>,
    pub mail_ids: Vec<Uuid>,
    /// In full: an admin deciding whether resending can work has to read the address.
    pub addresses: Vec<DbSecret>,
}

pub async fn get_stale_unclaimed(
    conn: &mut PgConnection,
    min_mail_count: i64,
    limit: i64,
) -> ModelResult<Vec<StaleUnclaimedLinkingMails>> {
    let res = sqlx::query_as!(
        StaleUnclaimedLinkingMails,
        r#"
SELECT e.student_number AS "student_number!",
  e.sisu_person_id AS "sisu_person_id!",
  e.course_id AS "course_id!",
  c.name AS "course_name!",
  COUNT(*) AS "mail_count!",
  MIN(e.sent_at) AS "first_sent_at!",
  MAX(e.sent_at) AS "last_sent_at!",
  ARRAY_AGG(
    e.id
    ORDER BY e.sent_at
  ) AS "mail_ids!",
  ARRAY_AGG(
    e.emailed_to
    ORDER BY e.sent_at
  ) AS "addresses!: Vec<DbSecret>"
FROM credit_registration_account_linking_emails e
  JOIN courses c ON c.id = e.course_id AND c.deleted_at IS NULL
WHERE e.deleted_at IS NULL
  AND NOT EXISTS (
    SELECT 1
    FROM verified_student_numbers vsn
    WHERE vsn.sisu_person_id = e.sisu_person_id
      AND vsn.deleted_at IS NULL
  )
GROUP BY e.student_number,
  e.sisu_person_id,
  e.course_id,
  c.name
HAVING COUNT(*) >= $1
ORDER BY MAX(e.sent_at) DESC
LIMIT $2
        "#,
        min_mail_count,
        limit,
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

/// Lets a rate-cap override or an admin resend mail the same address again.
pub async fn soft_delete(conn: &mut PgConnection, id: Uuid) -> ModelResult<()> {
    soft_delete_batch(conn, std::slice::from_ref(&id)).await
}

/// Batch form of [`soft_delete`]: one `UPDATE` for every row instead of one per id.
pub async fn soft_delete_batch(conn: &mut PgConnection, ids: &[Uuid]) -> ModelResult<()> {
    if ids.is_empty() {
        return Ok(());
    }
    sqlx::query!(
        r#"
UPDATE credit_registration_account_linking_emails
SET deleted_at = now()
WHERE id = ANY($1)
  AND deleted_at IS NULL
        "#,
        ids
    )
    .execute(conn)
    .await?;
    Ok(())
}

/// Voids the unused links in the mails about the person behind a verified link, and marks those
/// mails replaced, so that once unlinked the person can be mailed again. The mails still count
/// against the caps.
pub async fn retire_unused_for_link(
    conn: &mut PgConnection,
    verified_student_number_id: Uuid,
) -> ModelResult<()> {
    sqlx::query!(
        r#"
WITH retired AS (
  UPDATE credit_registration_account_linking_emails e
  SET replaced_at = now()
  FROM verified_student_numbers vsn
  WHERE vsn.id = $1
    AND e.deleted_at IS NULL
    AND e.replaced_at IS NULL
    AND (
      e.sisu_person_id = vsn.sisu_person_id
      OR e.student_number = vsn.student_number
    )
    AND NOT EXISTS (
      SELECT 1
      FROM student_number_verification_tokens t
      WHERE t.id = e.student_number_verification_token_id
        AND t.used_at IS NOT NULL
    )
  RETURNING e.student_number_verification_token_id
)
UPDATE student_number_verification_tokens t
SET deleted_at = now()
FROM retired
WHERE t.id = retired.student_number_verification_token_id
  AND t.deleted_at IS NULL
        "#,
        verified_student_number_id,
    )
    .execute(conn)
    .await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::email_deliveries::insert_email_delivery_to_address;
    use crate::email_templates::{EmailTemplateNew, EmailTemplateType, insert_email_template};
    use crate::library::credit_registration::account_linking::{
        DiscoveredPerson, claim_linking_mail,
    };
    use crate::test_helper::*;

    async fn claim_a_mail(conn: &mut PgConnection, course_id: Uuid) -> Uuid {
        claim_linking_mail(
            conn,
            &DiscoveredPerson {
                sisu_person_id: "hy-hlo-1".to_string().into(),
                student_number: "012345678".to_string().into(),
                first_names: Some("Aada".to_string().into()),
                last_name: Some("Virtanen".to_string().into()),
                course_id,
                address: Some(DbSecret::new("aada@example.com")),
            },
        )
        .await
        .unwrap();
        get_by_sisu_person_id(conn, "hy-hlo-1")
            .await
            .unwrap()
            .pop()
            .expect("the claim wrote a slot")
            .id
    }

    async fn seed_template(conn: &mut PgConnection) -> Uuid {
        insert_email_template(
            conn,
            None,
            EmailTemplateNew {
                template_type: EmailTemplateType::CreditRegistrationAccountLinking,
                language: Some("en".to_string()),
                content: Some(serde_json::json!([])),
                subject: Some("Link your student number".to_string()),
            },
            None,
        )
        .await
        .unwrap()
        .id
    }

    /// The property the two-phase split rests on: without it a restart between the claim and the
    /// queueing would mail the same address again.
    #[tokio::test]
    async fn a_claimed_slot_leaves_the_queue_once_its_delivery_exists() {
        insert_data!(:tx, :user, :org, :course);
        let slot = claim_a_mail(tx.as_mut(), course).await;
        let template = seed_template(tx.as_mut()).await;

        let claimed = claim_unqueued(tx.as_mut(), 10, None).await.unwrap();
        assert_eq!(claimed.len(), 1);
        assert_eq!(claimed[0].id, slot);
        assert_eq!(claimed[0].emailed_to.expose_secret(), "aada@example.com");

        let delivery = insert_email_delivery_to_address(
            tx.as_mut(),
            claimed[0].emailed_to.expose_secret(),
            template,
            &serde_json::json!({ "NAME": "Aada" }),
        )
        .await
        .unwrap();
        set_email_delivery_id(tx.as_mut(), slot, delivery)
            .await
            .unwrap();

        assert!(
            claim_unqueued(tx.as_mut(), 10, None)
                .await
                .unwrap()
                .is_empty()
        );
    }

    #[tokio::test]
    async fn a_slot_with_no_delivery_yet_reports_as_queued() {
        insert_data!(:tx, :user, :org, :course);
        let slot = claim_a_mail(tx.as_mut(), course).await;
        let reports = get_send_status_reports(tx.as_mut(), &[slot]).await.unwrap();
        assert_eq!(
            reports.get(&slot).map(|report| report.email_send_status),
            Some(EmailSendStatus::Queued)
        );
    }
}

/// A linking email whose link can still be used.
#[derive(Debug, Clone)]
pub struct UnusedLinkingLink {
    pub id: Uuid,
    pub course_id: Uuid,
    pub course_name: String,
    /// The course code of the email's course, where its modules agree on one.
    pub uh_course_code: Option<String>,
    pub emailed_to: DbSecret,
    pub claimed_at: DateTime<Utc>,
    pub expires_at: DateTime<Utc>,
}

/// Linking emails whose link can still be used, oldest first.
pub async fn get_unused_links(
    conn: &mut PgConnection,
    limit: i64,
) -> ModelResult<Vec<UnusedLinkingLink>> {
    let res = sqlx::query_as!(
        UnusedLinkingLink,
        r#"
SELECT e.id,
  e.course_id,
  c.name AS course_name,
  (
    SELECT MIN(TRIM(cm.uh_course_code))
    FROM course_modules cm
    WHERE cm.course_id = e.course_id
      AND cm.deleted_at IS NULL
      AND TRIM(COALESCE(cm.uh_course_code, '')) <> ''
  ) AS "uh_course_code?",
  e.emailed_to,
  e.sent_at AS claimed_at,
  t.expires_at
FROM credit_registration_account_linking_emails e
  JOIN courses c ON c.id = e.course_id AND c.deleted_at IS NULL
  JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
WHERE e.deleted_at IS NULL
  AND e.replaced_at IS NULL
  AND is_usable_verification_token(t)
ORDER BY e.sent_at,
  e.id
LIMIT $1
        "#,
        limit,
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

/// How many links can still be used per course code, over every course with a module on the code.
pub async fn count_unused_links_by_course_code(
    conn: &mut PgConnection,
) -> ModelResult<Vec<(String, i64)>> {
    let rows = sqlx::query!(
        r#"
SELECT TRIM(cm.uh_course_code) AS "course_code!",
  COUNT(DISTINCT e.id) AS "count!"
FROM credit_registration_account_linking_emails e
  JOIN courses c ON c.id = e.course_id AND c.deleted_at IS NULL
  JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
  JOIN course_modules cm ON cm.course_id = e.course_id
  AND cm.deleted_at IS NULL
  AND TRIM(COALESCE(cm.uh_course_code, '')) <> ''
WHERE e.deleted_at IS NULL
  AND e.replaced_at IS NULL
  AND is_usable_verification_token(t)
GROUP BY TRIM(cm.uh_course_code)
        "#,
    )
    .fetch_all(conn)
    .await?;
    Ok(rows
        .into_iter()
        .map(|row| (row.course_code, row.count))
        .collect())
}

/// Which linking emails a [`get_page`] lists, by what became of their link and their send.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum LinkingEmailListStatus {
    /// Carries a link that can still be used.
    Unused,
    Used,
    /// The link expired or was retired before anyone used it.
    Expired,
    /// Not handed to the mail server yet, or the send failed.
    NotSent,
}

impl LinkingEmailListStatus {
    /// The status name the `get_page` query matches on.
    fn as_sql(self) -> &'static str {
        match self {
            Self::Unused => "unused",
            Self::Used => "used",
            Self::Expired => "expired",
            Self::NotSent => "not_sent",
        }
    }
}

/// One linking email as the admin's searchable list shows it.
#[derive(Debug, Clone)]
pub struct LinkingEmailListRow {
    pub id: Uuid,
    pub course_id: Uuid,
    pub course_name: String,
    pub emailed_to: DbSecret,
    pub student_number: DbSecret,
    pub first_names: Option<DbSecret>,
    pub last_name: Option<DbSecret>,
    pub claimed_at: DateTime<Utc>,
    /// When the `link-emails` phase queued it for the email worker; `None` until then.
    pub queued_at: Option<DateTime<Utc>>,
    /// `None` until queued.
    pub send_status: Option<EmailSendStatusReport>,
    /// The newest delivery error's message, however transient.
    pub last_error_message: Option<String>,
    pub link_state: LinkingMailLinkState,
    /// `None` for a mail sent without a link.
    pub link_expires_at: Option<DateTime<Utc>>,
    pub link_used_at: Option<DateTime<Utc>>,
    pub claimed_by_user_id: Option<Uuid>,
    pub claimed_by_email: Option<String>,
    pub total_count: i64,
}

/// A page of linking emails on live courses, newest first, including ones a later mail replaced.
///
/// `address` matches any part of the recipient address, ignoring case, and overrides `status`, so a
/// search for a student finds their mail whatever became of it.
pub async fn get_page(
    conn: &mut PgConnection,
    status: Option<LinkingEmailListStatus>,
    address: Option<&str>,
    limit: i64,
    offset: i64,
) -> ModelResult<Vec<LinkingEmailListRow>> {
    let address_pattern = address.map(|address| {
        format!(
            "%{}%",
            crate::library::students_view::escape_like_pattern(address)
        )
    });
    let rows = sqlx::query!(
        r#"
WITH page AS (
  SELECT e.id,
    e.sent_at,
    c.name AS course_name,
    COUNT(*) OVER () AS total_count
  FROM credit_registration_account_linking_emails e
    JOIN courses c ON c.id = e.course_id
    AND c.deleted_at IS NULL
    LEFT JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
    LEFT JOIN email_deliveries ed ON ed.id = e.email_delivery_id
    AND ed.deleted_at IS NULL
  WHERE e.deleted_at IS NULL
    AND (
      $2::text IS NULL
      OR e.emailed_to ILIKE $2 ESCAPE '\'
    )
    AND (
      $1::text IS NULL
      OR (
        $1 = 'unused'
        AND is_usable_verification_token(t)
      )
      OR (
        $1 = 'used'
        AND t.used_at IS NOT NULL
      )
      OR (
        $1 = 'expired'
        AND e.student_number_verification_token_id IS NOT NULL
        AND t.used_at IS NULL
        AND NOT is_usable_verification_token(t)
      )
      OR (
        $1 = 'not_sent'
        AND ed.sent IS DISTINCT FROM TRUE
      )
    )
  ORDER BY e.sent_at DESC,
    e.id
  LIMIT $3 OFFSET $4
)
SELECT e.id,
  e.course_id,
  page.course_name AS "course_name!",
  e.emailed_to,
  e.student_number,
  t.first_names AS "first_names?",
  t.last_name AS "last_name?",
  e.sent_at AS claimed_at,
  ed.created_at AS "queued_at?",
  ed.sent AS "delivery_sent?",
  ed.retryable AS "retryable?",
  ed.retry_count AS "retry_count?",
  ed.next_retry_at AS "next_retry_at?",
  ed.first_failed_at AS "first_failed_at?",
  ed.last_attempt_at AS "last_attempt_at?",
  latest_error.error_code AS "error_code?",
  latest_error.error_message AS "error_message?",
  latest_error.is_transient AS "is_transient?",
  (
    e.student_number_verification_token_id IS NULL
    OR is_usable_verification_token(t)
  ) AS "is_usable!",
  t.expires_at AS "link_expires_at?",
  t.used_at AS "link_used_at?",
  t.claimed_by_user_id AS "claimed_by_user_id?",
  claimer.email AS "claimed_by_email?",
  page.total_count AS "total_count!"
FROM page
  JOIN credit_registration_account_linking_emails e ON e.id = page.id
  LEFT JOIN student_number_verification_tokens t ON t.id = e.student_number_verification_token_id
  LEFT JOIN user_details claimer ON claimer.user_id = t.claimed_by_user_id
  LEFT JOIN email_deliveries ed ON ed.id = e.email_delivery_id
  AND ed.deleted_at IS NULL
  LEFT JOIN LATERAL (
    SELECT ede.error_code,
      ede.error_message,
      ede.is_transient
    FROM email_delivery_errors ede
    WHERE ede.email_delivery_id = ed.id
      AND ede.deleted_at IS NULL
    ORDER BY ede.attempt DESC,
      ede.created_at DESC
    LIMIT 1
  ) latest_error ON TRUE
ORDER BY page.sent_at DESC,
  page.id
        "#,
        status
            .filter(|_| address.is_none())
            .map(LinkingEmailListStatus::as_sql),
        address_pattern,
        limit,
        offset,
    )
    .fetch_all(conn)
    .await?;
    let now = Utc::now();
    Ok(rows
        .into_iter()
        .map(|row| {
            let send_status = row.delivery_sent.map(|sent| {
                derive_email_send_status(
                    &EmailSendStatusFacts {
                        sent,
                        retryable: row.retryable.unwrap_or(false),
                        retry_count: row.retry_count.unwrap_or(0),
                        next_retry_at: row.next_retry_at,
                        first_failed_at: row.first_failed_at,
                        last_attempt_at: row.last_attempt_at,
                        failure_code: row.error_code,
                        failure_is_transient: row.is_transient,
                    },
                    now,
                )
            });
            LinkingEmailListRow {
                id: row.id,
                course_id: row.course_id,
                course_name: row.course_name,
                emailed_to: row.emailed_to,
                student_number: row.student_number,
                first_names: row.first_names,
                last_name: row.last_name,
                claimed_at: row.claimed_at,
                queued_at: row.queued_at,
                send_status,
                last_error_message: row.error_message,
                link_state: LinkingMailLinkState::of(row.is_usable, row.link_used_at.is_some()),
                link_expires_at: row.link_expires_at,
                link_used_at: row.link_used_at,
                claimed_by_user_id: row.claimed_by_user_id,
                claimed_by_email: row.claimed_by_email,
                total_count: row.total_count,
            }
        })
        .collect())
}