headless_lms_models/credit_registrations/

metrics.rs

//! The dashboard's and the health alerts' counts over the whole ledger.

use super::attention::{AttentionReason, AttentionRegistration};
use super::state::{CreditRegistrationErrorCode, CreditRegistrationState};
use crate::credit_registration_enrolment_routes::CreditRegistrationEnrolmentRoute;
use crate::library::credit_registration::PendingReasonCounts;
use crate::prelude::*;
use chrono::TimeDelta;
use std::collections::HashMap;
use utoipa::ToSchema;

/// Live rows per state, for the dashboard funnel. Superseded attempts are excluded, as in the
/// per-course sibling, or a course that regrades counts every student twice.
pub async fn count_by_state(
    conn: &mut PgConnection,
) -> ModelResult<Vec<(CreditRegistrationState, i64)>> {
    let rows = sqlx::query!(
        r#"
SELECT state,
  COUNT(*) AS "count!"
FROM credit_registrations
WHERE superseded_by_id IS NULL
  AND deleted_at IS NULL
GROUP BY state
        "#,
    )
    .fetch_all(conn)
    .await?;
    Ok(rows.into_iter().map(|r| (r.state, r.count)).collect())
}

/// Live rows parked in `no_usable_enrolment` that an unscoped resolve-enrolments claim would take
/// for an enrolment check now; the rest wait for their schedule, their module, or a lookup already
/// out. Leaves out the claim's one-row-per-student-and-module hold, which only defers a row.
///
/// Shares its filters with the unscoped `claim` and with the released-check test of
/// [`pull_forward_batched_checks`](crate::library::credit_registration::enrolment_checks::pull_forward_batched_checks);
/// change all three together, or the queue depth counts rows no claim takes.
pub async fn count_due_enrolment_checks(conn: &mut PgConnection) -> ModelResult<i64> {
    let count = sqlx::query_scalar!(
        r#"
SELECT COUNT(*) AS "count!"
FROM credit_registrations cr
  JOIN credit_registration_active_course_modules acm ON acm.course_module_id = cr.course_module_id
  JOIN course_module_completions cmc ON cmc.id = cr.course_module_completion_id
WHERE cr.state = 'no_usable_enrolment'
  AND cr.next_attempt_at <= now()
  AND cr.superseded_by_id IS NULL
  AND cr.deleted_at IS NULL
  AND (
    cmc.register_credits_via_suotar
    OR cr.submitted_at IS NOT NULL
  )
  AND (
    cr.enrolment_check_claimed_until IS NULL
    OR cr.enrolment_check_claimed_until <= now()
  )
  AND NOT EXISTS (
    SELECT 1
    FROM credit_registration_test_exclusive_holds h
    WHERE h.user_id = cr.user_id
      AND (
        h.course_id IS NULL
        OR h.course_id = cr.course_id
      )
      AND h.held_until > now()
  )
        "#,
    )
    .fetch_one(conn)
    .await?;
    Ok(count)
}

/// Live `pending` rows per blocker. Derived from `credit_registration_preconditions`, so it cannot
/// disagree with what the recompute is waiting for or with what the student is shown.
pub async fn count_pending_by_reason(conn: &mut PgConnection) -> ModelResult<PendingReasonCounts> {
    let row = sqlx::query!(
        r#"
SELECT COUNT(*) FILTER (
    WHERE NOT p.completion_eligible
  ) AS "completion_count!",
  COUNT(*) FILTER (
    WHERE p.completion_eligible
      AND NOT p.has_verified_student_number
  ) AS "student_number_count!",
  COUNT(*) FILTER (
    WHERE p.completion_eligible
      AND p.has_verified_student_number
      AND NOT p.course_code_allowed
  ) AS "course_code_count!"
FROM credit_registrations cr
  JOIN credit_registration_preconditions p ON p.credit_registration_id = cr.id
WHERE cr.state = 'pending'
  AND cr.superseded_by_id IS NULL
  AND cr.deleted_at IS NULL
        "#,
    )
    .fetch_one(conn)
    .await?;
    Ok(PendingReasonCounts {
        completion_count: row.completion_count,
        student_number_count: row.student_number_count,
        course_code_count: row.course_code_count,
    })
}

/// A `pending` registration waiting for a student number, with what its student has done about it.
#[derive(Debug, Clone, PartialEq)]
pub struct WaitingForStudentNumber {
    pub credit_registration_id: Uuid,
    pub user_id: Uuid,
    pub email: Option<String>,
    pub first_name: Option<String>,
    pub last_name: Option<String>,
    pub course_id: Uuid,
    pub course_name: String,
    pub course_module_name: Option<String>,
    pub uh_course_code: Option<String>,
    pub completion_date: DateTime<Utc>,
    pub last_visited_at: Option<DateTime<Utc>>,
    /// The last "I have enrolled" press; `None` when the latest check request came from someone else.
    pub last_check_requested_at: Option<DateTime<Utc>>,
    /// The student's standing "I have enrolled" press, which is what makes them Pressed.
    pub pressed_at: Option<DateTime<Utc>>,
}

/// The rows behind [`PendingReasonCounts::student_number_count`] completed on or after `since`,
/// longest waiting first, with how many there are in all.
pub async fn get_waiting_for_student_number(
    conn: &mut PgConnection,
    since: Option<DateTime<Utc>>,
    limit: i64,
) -> ModelResult<(Vec<WaitingForStudentNumber>, i64)> {
    let rows = sqlx::query!(
        r#"
SELECT cr.id AS credit_registration_id,
  cr.user_id,
  ud.email AS "email?",
  ud.first_name AS "first_name?",
  ud.last_name AS "last_name?",
  cr.course_id,
  c.name AS course_name,
  cm.name AS "course_module_name?",
  cm.uh_course_code AS "uh_course_code?",
  cmc.completion_date,
  sig.last_visited_at AS "last_visited_at?",
  CASE
    WHEN sig.check_request_source = 'student_request' THEN sig.last_check_requested_at
  END AS "last_check_requested_at?",
  route.enrolment_confirmed_at AS "pressed_at?",
  COUNT(*) OVER () AS "total!"
FROM credit_registrations cr
  JOIN credit_registration_preconditions p ON p.credit_registration_id = cr.id
  JOIN course_module_completions cmc ON cmc.id = cr.course_module_completion_id
  AND cmc.deleted_at IS NULL
  JOIN courses c ON c.id = cr.course_id
  AND c.deleted_at IS NULL
  JOIN course_modules cm ON cm.id = cr.course_module_id
  AND cm.deleted_at IS NULL
  LEFT JOIN user_details ud ON ud.user_id = cr.user_id
  LEFT JOIN credit_registration_enrolment_check_signals sig ON sig.course_module_completion_id = cr.course_module_completion_id
  AND sig.deleted_at IS NULL
  LEFT JOIN credit_registration_enrolment_routes route ON route.course_module_completion_id = cr.course_module_completion_id
  AND route.deleted_at IS NULL
WHERE cr.state = 'pending'
  AND cr.superseded_by_id IS NULL
  AND cr.deleted_at IS NULL
  AND p.completion_eligible
  AND NOT p.has_verified_student_number
  AND ($1::timestamptz IS NULL OR cmc.completion_date >= $1)
ORDER BY cmc.completion_date,
  cr.id
LIMIT $2
        "#,
        since,
        limit,
    )
    .fetch_all(conn)
    .await?;
    let total = rows.first().map_or(0, |row| row.total);
    Ok((
        rows.into_iter()
            .map(|row| WaitingForStudentNumber {
                credit_registration_id: row.credit_registration_id,
                user_id: row.user_id,
                email: row.email,
                first_name: row.first_name,
                last_name: row.last_name,
                course_id: row.course_id,
                course_name: row.course_name,
                course_module_name: row.course_module_name,
                uh_course_code: row.uh_course_code,
                completion_date: row.completion_date,
                last_visited_at: row.last_visited_at,
                last_check_requested_at: row.last_check_requested_at,
                pressed_at: row.pressed_at,
            })
            .collect(),
        total,
    ))
}

/// A student waiting for a student number who pressed "I have enrolled", with what has happened on
/// their course code since.
#[derive(Debug, Clone, PartialEq)]
pub struct WaitingPresser {
    pub credit_registration_id: Uuid,
    pub user_id: Uuid,
    pub email: Option<String>,
    pub first_name: Option<String>,
    pub last_name: Option<String>,
    pub course_id: Uuid,
    pub course_name: String,
    pub course_module_id: Uuid,
    pub course_module_name: Option<String>,
    /// The module's code now, trimmed.
    pub uh_course_code: Option<String>,
    pub completion_date: DateTime<Utc>,
    pub pressed_at: DateTime<Utc>,
    pub enrolment_route: CreditRegistrationEnrolmentRoute,
    /// Linking emails claimed on any course sharing the code since the press. Not attributable to
    /// this student: an email is tied to an account only once its link is used.
    pub linking_emails_on_code_since_press: i64,
    pub last_linking_email_on_code_at: Option<DateTime<Utc>>,
}

/// Students waiting for a student number who pressed "I have enrolled" and completed on or after
/// `since`, longest waiting first. `credit_registration_id` narrows to one row.
pub async fn get_waiting_pressers(
    conn: &mut PgConnection,
    since: Option<DateTime<Utc>>,
    credit_registration_id: Option<Uuid>,
) -> ModelResult<Vec<WaitingPresser>> {
    let res = sqlx::query_as!(
        WaitingPresser,
        r#"
SELECT cr.id AS credit_registration_id,
  cr.user_id,
  ud.email AS "email?",
  ud.first_name AS "first_name?",
  ud.last_name AS "last_name?",
  cr.course_id,
  c.name AS course_name,
  cr.course_module_id,
  cm.name AS "course_module_name?",
  NULLIF(TRIM(cm.uh_course_code), '') AS "uh_course_code?",
  cmc.completion_date,
  route.enrolment_confirmed_at AS "pressed_at!",
  route.route AS "enrolment_route: CreditRegistrationEnrolmentRoute",
  emails.count AS "linking_emails_on_code_since_press!",
  emails.last_at AS "last_linking_email_on_code_at?"
FROM credit_registrations cr
  JOIN credit_registration_preconditions p ON p.credit_registration_id = cr.id
  JOIN course_module_completions cmc ON cmc.id = cr.course_module_completion_id
  AND cmc.deleted_at IS NULL
  JOIN courses c ON c.id = cr.course_id
  AND c.deleted_at IS NULL
  JOIN course_modules cm ON cm.id = cr.course_module_id
  AND cm.deleted_at IS NULL
  JOIN credit_registration_enrolment_routes route ON route.course_module_completion_id = cr.course_module_completion_id
  AND route.deleted_at IS NULL
  AND route.enrolment_confirmed_at IS NOT NULL
  LEFT JOIN user_details ud ON ud.user_id = cr.user_id
  CROSS JOIN LATERAL (
    SELECT COUNT(*) AS count,
      MAX(e.sent_at) AS last_at
    FROM credit_registration_account_linking_emails e
      JOIN course_modules code_module ON code_module.course_id = e.course_id
      AND code_module.deleted_at IS NULL
    WHERE TRIM(code_module.uh_course_code) = TRIM(cm.uh_course_code)
      AND e.sent_at >= route.enrolment_confirmed_at
      AND e.deleted_at IS NULL
  ) emails
WHERE cr.state = 'pending'
  AND cr.superseded_by_id IS NULL
  AND cr.deleted_at IS NULL
  AND p.completion_eligible
  AND NOT p.has_verified_student_number
  AND ($1::timestamptz IS NULL OR cmc.completion_date >= $1)
  AND ($2::uuid IS NULL OR cr.id = $2)
ORDER BY route.enrolment_confirmed_at,
  cr.id
        "#,
        since,
        credit_registration_id,
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

/// Live rows carrying an error code, split by whether the pipeline is still working on them.
#[derive(Debug, Clone, PartialEq)]
pub struct CreditRegistrationErrorCodeCount {
    pub error_code: CreditRegistrationErrorCode,
    /// Every live row carrying the code: what the Registrations list filtered by it shows.
    pub live_count: i64,
    pub in_flight_count: i64,
    pub terminal_failure_count: i64,
}

/// The error-code breakdown the Overview shows.
pub async fn count_by_error_code(
    conn: &mut PgConnection,
) -> ModelResult<Vec<CreditRegistrationErrorCodeCount>> {
    let rows = sqlx::query!(
        r#"
SELECT error_code AS "error_code!",
  COUNT(*) AS "live_count!",
  COUNT(*) FILTER (WHERE terminal_at IS NULL) AS "in_flight_count!",
  COUNT(*) FILTER (
    WHERE state = ANY($1::credit_registration_state [])
  ) AS "terminal_failure_count!"
FROM credit_registrations
WHERE error_code IS NOT NULL
  AND superseded_by_id IS NULL
  AND deleted_at IS NULL
GROUP BY error_code
ORDER BY COUNT(*) DESC
        "#,
        &CreditRegistrationState::HARD_FAILURE_STATES as &[CreditRegistrationState],
    )
    .fetch_all(conn)
    .await?;
    Ok(rows
        .into_iter()
        .map(|row| CreditRegistrationErrorCodeCount {
            error_code: row.error_code,
            live_count: row.live_count,
            in_flight_count: row.in_flight_count,
            terminal_failure_count: row.terminal_failure_count,
        })
        .collect())
}

/// The row that has been waiting longest for the pipeline to do something with it.
#[derive(Debug, Clone, PartialEq)]
pub struct OldestNonTerminalRegistration {
    pub id: Uuid,
    pub state: CreditRegistrationState,
    pub state_changed_at: DateTime<Utc>,
}

pub async fn get_oldest_non_terminal(
    conn: &mut PgConnection,
) -> ModelResult<Option<OldestNonTerminalRegistration>> {
    let row = sqlx::query_as!(
        OldestNonTerminalRegistration,
        r#"
SELECT id,
  state,
  state_changed_at
FROM credit_registrations
WHERE terminal_at IS NULL
  AND superseded_by_id IS NULL
  AND deleted_at IS NULL
ORDER BY state_changed_at
LIMIT 1
        "#,
    )
    .fetch_optional(conn)
    .await?;
    Ok(row)
}

/// One day of terminal outcomes, for the throughput series.
#[derive(Debug, Clone, PartialEq)]
pub struct CreditRegistrationThroughputDay {
    pub day: DateTime<Utc>,
    pub registered_count: i64,
    pub other_success_count: i64,
    pub failed_count: i64,
}

/// Daily terminal outcomes over the window. Withdrawn rows are in no column: they are neither a
/// success nor a failure.
pub async fn get_throughput_by_day(
    conn: &mut PgConnection,
    since: DateTime<Utc>,
) -> ModelResult<Vec<CreditRegistrationThroughputDay>> {
    let rows = sqlx::query_as!(
        CreditRegistrationThroughputDay,
        r#"
SELECT DATE_TRUNC('day', terminal_at) AS "day!",
  COUNT(*) FILTER (WHERE state = 'registered') AS "registered_count!",
  COUNT(*) FILTER (
    WHERE state = ANY($2::credit_registration_state [])
  ) AS "other_success_count!",
  COUNT(*) FILTER (WHERE state = 'failed_permanent') AS "failed_count!"
FROM credit_registrations
WHERE terminal_at >= $1
  AND superseded_by_id IS NULL
  AND deleted_at IS NULL
GROUP BY 1
ORDER BY 1
        "#,
        since,
        &CreditRegistrationState::OTHER_SUCCESS_STATES as &[CreditRegistrationState],
    )
    .fetch_all(conn)
    .await?;
    Ok(rows)
}

/// What the pipeline finished in a window.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct TerminalOutcomeTotals {
    /// `registered`, `duplicate` and `not_improved`.
    pub success_count: i64,
    /// The subset we put in the registry ourselves.
    pub registered_count: i64,
    pub failed_permanent_count: i64,
    pub cancelled_count: i64,
    /// The denominator of the success rate.
    pub total_count: i64,
}

pub async fn count_terminal_outcomes_since(
    conn: &mut PgConnection,
    since: DateTime<Utc>,
) -> ModelResult<TerminalOutcomeTotals> {
    let res = sqlx::query_as!(
        TerminalOutcomeTotals,
        r#"
SELECT COUNT(*) FILTER (
    WHERE state = ANY($2::credit_registration_state [])
  ) AS "success_count!",
  COUNT(*) FILTER (WHERE state = 'registered') AS "registered_count!",
  COUNT(*) FILTER (WHERE state = 'failed_permanent') AS "failed_permanent_count!",
  COUNT(*) FILTER (WHERE state = 'cancelled') AS "cancelled_count!",
  COUNT(*) AS "total_count!"
FROM credit_registrations
WHERE terminal_at >= $1
  AND superseded_by_id IS NULL
  AND deleted_at IS NULL
        "#,
        since,
        &CreditRegistrationState::SUCCESS_STATES as &[CreditRegistrationState],
    )
    .fetch_one(conn)
    .await?;
    Ok(res)
}

/// Live rows that entered one state within the window. `misregistered` is not terminal, so
/// `terminal_at` cannot answer this.
pub async fn count_entered_state_since(
    conn: &mut PgConnection,
    state: CreditRegistrationState,
    since: DateTime<Utc>,
) -> ModelResult<i64> {
    let count = sqlx::query_scalar!(
        r#"
SELECT COUNT(*) AS "count!"
FROM credit_registrations
WHERE state = $1
  AND state_changed_at >= $2
  AND superseded_by_id IS NULL
  AND deleted_at IS NULL
        "#,
        state as CreditRegistrationState,
        since,
    )
    .fetch_one(conn)
    .await?;
    Ok(count)
}

/// How long registration took, in seconds, for rows that reached `registered` in a window.
#[derive(Debug, Clone, PartialEq)]
pub struct RegistrationLatency {
    pub registered_count: i64,
    /// `terminal_at - created_at`: the student's wait, most of which is theirs to end.
    pub p50_end_to_end_secs: Option<i64>,
    pub p95_end_to_end_secs: Option<i64>,
    /// `registered_at - submitted_at`: how long the study registry took, which is the number to
    /// quote at them.
    pub p50_confirmation_secs: Option<i64>,
    pub p95_confirmation_secs: Option<i64>,
}

pub async fn get_registration_latency_between(
    conn: &mut PgConnection,
    from: DateTime<Utc>,
    to: DateTime<Utc>,
) -> ModelResult<RegistrationLatency> {
    let res = sqlx::query_as!(
        RegistrationLatency,
        r#"
SELECT COUNT(*) AS "registered_count!",
  CEIL(
    EXTRACT(
      EPOCH
      FROM PERCENTILE_DISC(0.5) WITHIN GROUP (
          ORDER BY terminal_at - created_at
        )
    )
  )::bigint AS "p50_end_to_end_secs",
  CEIL(
    EXTRACT(
      EPOCH
      FROM PERCENTILE_DISC(0.95) WITHIN GROUP (
          ORDER BY terminal_at - created_at
        )
    )
  )::bigint AS "p95_end_to_end_secs",
  CEIL(
    EXTRACT(
      EPOCH
      FROM PERCENTILE_DISC(0.5) WITHIN GROUP (
          ORDER BY registered_at - submitted_at
        )
    )
  )::bigint AS "p50_confirmation_secs",
  CEIL(
    EXTRACT(
      EPOCH
      FROM PERCENTILE_DISC(0.95) WITHIN GROUP (
          ORDER BY registered_at - submitted_at
        )
    )
  )::bigint AS "p95_confirmation_secs"
FROM credit_registrations
WHERE state = 'registered'
  AND terminal_at >= $1
  AND terminal_at < $2
  AND superseded_by_id IS NULL
  AND deleted_at IS NULL
        "#,
        from,
        to,
    )
    .fetch_one(conn)
    .await?;
    Ok(res)
}

/// Live volumes per course module, for the Courses tab's one row per module.
#[derive(Debug, Clone, PartialEq)]
pub struct ModuleRegistrationTotals {
    pub course_module_id: Uuid,
    pub total_count: i64,
    pub success_count: i64,
    /// Of `success_count`, the ones we registered.
    pub registered_count: i64,
    /// Of `success_count`, the ones Sisu already held: duplicate or not improved.
    pub already_in_sisu_count: i64,
    pub in_flight_count: i64,
    pub failed_count: i64,
    pub last_registered_at: Option<DateTime<Utc>>,
    /// The code most of the module's failing rows carry, which is usually the whole diagnosis.
    pub top_error_code: Option<CreditRegistrationErrorCode>,
}

/// `failed_count` is `failed_permanent` and `misregistered` only, so the columns do not add up to
/// the total by design.
pub async fn count_by_module(
    conn: &mut PgConnection,
) -> ModelResult<Vec<ModuleRegistrationTotals>> {
    let res = sqlx::query_as!(
        ModuleRegistrationTotals,
        r#"
SELECT cr.course_module_id,
  COUNT(*) AS "total_count!",
  COUNT(*) FILTER (
    WHERE cr.state = ANY($1::credit_registration_state [])
  ) AS "success_count!",
  COUNT(*) FILTER (WHERE cr.state = 'registered') AS "registered_count!",
  COUNT(*) FILTER (
    WHERE cr.state = ANY($3::credit_registration_state [])
  ) AS "already_in_sisu_count!",
  COUNT(*) FILTER (WHERE cr.terminal_at IS NULL) AS "in_flight_count!",
  COUNT(*) FILTER (
    WHERE cr.state = ANY($2::credit_registration_state [])
  ) AS "failed_count!",
  MAX(cr.registered_at) AS "last_registered_at",
  (
    SELECT inner_cr.error_code
    FROM credit_registrations inner_cr
    WHERE inner_cr.course_module_id = cr.course_module_id
      AND inner_cr.error_code IS NOT NULL
      AND inner_cr.superseded_by_id IS NULL
      AND inner_cr.deleted_at IS NULL
    GROUP BY inner_cr.error_code
    ORDER BY COUNT(*) DESC,
      inner_cr.error_code
    LIMIT 1
  ) AS "top_error_code?: CreditRegistrationErrorCode"
FROM credit_registrations cr
WHERE superseded_by_id IS NULL
  AND deleted_at IS NULL
GROUP BY cr.course_module_id
        "#,
        &CreditRegistrationState::SUCCESS_STATES as &[CreditRegistrationState],
        &CreditRegistrationState::HARD_FAILURE_STATES as &[CreditRegistrationState],
        &CreditRegistrationState::OTHER_SUCCESS_STATES as &[CreditRegistrationState],
    )
    .fetch_all(conn)
    .await?;
    Ok(res)
}

/// How long a row may sit in one state before it counts as stuck. Seconds, per state.
///
/// Also the wire payload the health endpoint reports as thresholds: field names are the
/// `stuck_*_secs` keys the frontend reads, so do not rename without updating it.
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, ToSchema)]
pub struct StuckThresholds {
    pub stuck_ready_to_submit_secs: i64,
    pub stuck_submitting_secs: i64,
    pub stuck_awaiting_verification_secs: i64,
    pub stuck_failed_retryable_secs: i64,
}

/// The longest `submissionPending` asks verify to wait before polling again.
const SUOTAR_PENDING_WAIT: TimeDelta = TimeDelta::days(1);

impl StuckThresholds {
    /// The thresholds every detector and alert uses.
    pub const CURRENT: Self = Self {
        stuck_ready_to_submit_secs: 2 * 60 * 60,
        stuck_submitting_secs: 90 * 60,
        stuck_awaiting_verification_secs: SUOTAR_PENDING_WAIT.num_seconds() + 2 * 60 * 60,
        stuck_failed_retryable_secs: 3 * 24 * 60 * 60,
    };

    /// The threshold for `state`, or `None` for a state that waits on a student or a person.
    pub fn seconds_for(&self, state: CreditRegistrationState) -> Option<i64> {
        let (states, seconds) = self.state_seconds_arrays();
        states
            .iter()
            .position(|candidate| *candidate == state)
            .map(|index| seconds[index] as i64)
    }

    /// The four states this covers, paired with their threshold in seconds, for a query to `UNNEST`
    /// into a state -> threshold lookup.
    pub(super) fn state_seconds_arrays(&self) -> ([CreditRegistrationState; 4], [f64; 4]) {
        (
            [
                CreditRegistrationState::ReadyToSubmit,
                CreditRegistrationState::Submitting,
                CreditRegistrationState::AwaitingVerification,
                CreditRegistrationState::FailedRetryable,
            ],
            [
                self.stuck_ready_to_submit_secs as f64,
                self.stuck_submitting_secs as f64,
                self.stuck_awaiting_verification_secs as f64,
                self.stuck_failed_retryable_secs as f64,
            ],
        )
    }
}

const _: () = assert!(
    StuckThresholds::CURRENT.stuck_failed_retryable_secs
        < crate::library::credit_registration::backoff::SUBMIT_MAX_RETRY_AGE.num_seconds(),
    "a row must be considered stuck before backoff gives up retrying it"
);
const _: () = assert!(
    StuckThresholds::CURRENT.stuck_submitting_secs
        > crate::library::credit_registration::backoff::SUBMITTING_RECOVERY_GRACE.num_seconds(),
    "the stuck threshold must outlast the grace period that lets a submit recover on its own"
);

#[derive(Debug, Clone, PartialEq)]
pub struct StuckRegistrationCount {
    pub state: CreditRegistrationState,
    pub count: i64,
    /// Over three times the threshold, which is what makes the alert critical.
    pub severely_stuck_count: i64,
    pub oldest_state_changed_at: Option<DateTime<Utc>>,
}

/// Rows past their state's threshold, per state, for the stuck-registrations alert. Counted
/// whatever their Needs attention standing: a dismissal or another reason does not unstick a row.
pub fn count_stuck_in_state(
    rows: &[AttentionRegistration],
    thresholds: &StuckThresholds,
    now: DateTime<Utc>,
) -> Vec<StuckRegistrationCount> {
    let mut by_state: HashMap<CreditRegistrationState, StuckRegistrationCount> = HashMap::new();
    for row in rows
        .iter()
        .filter(|row| row.reasons.contains(&AttentionReason::StuckInState))
    {
        let severe = thresholds
            .seconds_for(row.state)
            .is_some_and(|secs| (now - row.state_changed_at).num_seconds() > secs * 3);
        let count = by_state.entry(row.state).or_insert(StuckRegistrationCount {
            state: row.state,
            count: 0,
            severely_stuck_count: 0,
            oldest_state_changed_at: None,
        });
        count.count += 1;
        count.severely_stuck_count += i64::from(severe);
        count.oldest_state_changed_at = Some(
            count
                .oldest_state_changed_at
                .map_or(row.state_changed_at, |oldest| {
                    oldest.min(row.state_changed_at)
                }),
        );
    }
    let mut counts: Vec<StuckRegistrationCount> = by_state.into_values().collect();
    counts.sort_by_key(|count| {
        CreditRegistrationState::ALL
            .iter()
            .position(|state| *state == count.state)
    });
    counts
}